dohertj2/lmxopcua
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge hw3-history (#655 delete dead driver-side history capability + client surfaces Bad HistoryRead)

Browse Source 
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-09 00:14:47 -04:00
parent 7c4cc31842 67059e479f
commit 862d8a756b
47 changed files with 289 additions and 1709 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Adapters/DefaultSessionAdapter.cs
+6 -4
View File
@@ -1,6 +1,7 @@
using Opc.Ua;
using Opc.Ua.Client;
using Serilog;
using ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers;
namespace ZB.MOM.WW.OtOpcUa.Client.Shared.Adapters;
@@ -172,8 +173,9 @@ internal sealed class DefaultSessionAdapter : ISessionAdapter
break;
var result = results[0];
if (StatusCode.IsBad(result.StatusCode))
break;
// A refused read (BadHistoryOperationUnsupported, BadUserAccessDenied, ...) must surface; only
// BadNoData is an empty result. Swallowing it printed "0 values returned" for a refusal.
HistoryReadStatus.ThrowIfRefused(result.StatusCode, nodeId);
if (result.HistoryData is ExtensionObject ext && ext.Body is HistoryData historyData)
allValues.AddRange(historyData.DataValues);
@@ -218,8 +220,8 @@ internal sealed class DefaultSessionAdapter : ISessionAdapter
if (results != null && results.Count > 0)
{
var result = results[0];
if (!StatusCode.IsBad(result.StatusCode) &&
result.HistoryData is ExtensionObject ext &&
HistoryReadStatus.ThrowIfRefused(result.StatusCode, nodeId);
if (result.HistoryData is ExtensionObject ext &&
ext.Body is HistoryData historyData)
allValues.AddRange(historyData.DataValues);
}
src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Helpers/HistoryReadStatus.cs
+24
View File
@@ -0,0 +1,24 @@
using Opc.Ua;
namespace ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers;
/// <summary>
/// Classifies the per-node <c>HistoryReadResult.StatusCode</c> a server returns for a HistoryRead.
/// A Bad per-node status (<c>BadHistoryOperationUnsupported</c>, <c>BadUserAccessDenied</c>,
/// <c>BadNodeIdUnknown</c>, ...) is a refused read, not an empty one, so it must reach the caller
/// instead of collapsing to "0 values returned". <c>BadNoData</c> is the one Bad code that genuinely
/// means "the window held nothing" and stays an empty result.
/// </summary>
internal static class HistoryReadStatus
{
/// <summary>Throws a <see cref="ServiceResultException"/> naming the node and status when the read was refused.</summary>
/// <param name="status">The per-node result status.</param>
/// <param name="nodeId">The node that was read, for the message.</param>
/// <exception cref="ServiceResultException"><paramref name="status"/> is Bad and not <c>BadNoData</c>.</exception>
public static void ThrowIfRefused(StatusCode status, NodeId nodeId)
{
if (!StatusCode.IsBad(status) || status.Code == StatusCodes.BadNoData) return;
throw new ServiceResultException(status.Code,
$"HistoryRead of {nodeId} failed: {StatusCodes.GetBrowseName(status.Code)} (0x{status.Code:X8}).");
}
}
src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/DriverCapability.cs
+2 -5
View File
@@ -5,10 +5,10 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
/// Each value corresponds to one method (or tightly-related method group) on the
/// <c>Core.Abstractions</c> capability interfaces (<see cref="IReadable"/>, <see cref="IWritable"/>,
/// <see cref="ITagDiscovery"/>, <see cref="ISubscribable"/>, <see cref="IHostConnectivityProbe"/>,
/// <see cref="IAlarmSource"/>, <see cref="IHistoryProvider"/>).
/// <see cref="IAlarmSource"/>).
/// </summary>
/// <remarks>
/// Per the per-capability retry policy in <c>docs/v2/plan.md</c>: Read / HistoryRead /
/// Per the per-capability retry policy in <c>docs/v2/plan.md</c>: Read /
/// Discover / Probe / AlarmSubscribe auto-retry; <see cref="Write"/> does NOT retry unless the
/// tag-definition carries <see cref="WriteIdempotentAttribute"/>. Alarm-acknowledge is treated
/// as a write for retry semantics (an alarm-ack is not idempotent at the plant-floor acknowledgement
@@ -37,9 +37,6 @@ public enum DriverCapability
/// <summary><see cref="IAlarmSource.AcknowledgeAsync"/>. Does NOT retry — ack is a write-shaped operation.</summary>
AlarmAcknowledge,
/// <summary><see cref="IHistoryProvider"/> reads (Raw/Processed/AtTime/Events). Retries by default.</summary>
HistoryRead,
/// <summary>
/// <c>IDriver.InitializeAsync</c> — connection establishment (R2-10 T10 closure: without this,
/// a driver that cannot connect contributes zero breaker input and the resilience panel stays
src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs
+92
View File
@@ -0,0 +1,92 @@
namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
// Shared history result contracts used by IHistorianDataSource and the historian gateway.
// Drivers carry no history capability (Gitea #655): HistoryRead is served server-side.
/// <summary>Result of a HistoryRead call.</summary>
/// <param name="Samples">Returned samples in chronological order.</param>
/// <param name="ContinuationPoint">Opaque token for the next call when more samples are available; null when complete.</param>
public sealed record HistoryReadResult(
IReadOnlyList<DataValueSnapshot> Samples,
byte[]? ContinuationPoint);
/// <summary>Aggregate function for processed history reads. Mirrors OPC UA Part 13 standard aggregates.</summary>
/// <remarks>
/// The first five members are the original set every historian backend serves. The remainder (#285) are
/// the further standard Part 13 aggregates a server-side HistoryRead can forward to a backend that
/// supports them (a gateway backend maps them to its own aggregate modes). A backend that cannot serve a
/// member must throw <see cref="HistoryAggregateNotSupportedException"/> — the server maps that to
/// <c>BadAggregateNotSupported</c> rather than a generic failure. <b>Append only</b>: the numeric values
/// are not persisted, but every <c>switch</c> over this enum must keep a total default.
/// </remarks>
public enum HistoryAggregateType
{
Average,
Minimum,
Maximum,
Total,
Count,
Interpolative,
TimeAverage,
Start,
End,
Delta,
Range,
MinimumActualTime,
MaximumActualTime,
StartBound,
EndBound,
DeltaBounds,
DurationGood,
DurationBad,
PercentGood,
PercentBad,
WorstQuality,
NumberOfTransitions,
AnnotationCount,
StandardDeviationPopulation,
VariancePopulation,
}
/// <summary>
/// Thrown by a history backend when it cannot serve the requested <see cref="HistoryAggregateType"/>.
/// The Server HistoryRead path translates it to <c>BadAggregateNotSupported</c> for that node only.
/// </summary>
public sealed class HistoryAggregateNotSupportedException : NotSupportedException
{
/// <summary>The aggregate the backend could not serve.</summary>
public HistoryAggregateType Aggregate { get; }
/// <summary>Creates the exception for <paramref name="aggregate"/>.</summary>
/// <param name="aggregate">The unsupported aggregate.</param>
public HistoryAggregateNotSupportedException(HistoryAggregateType aggregate)
: base($"The historian backend does not support the {aggregate} aggregate.")
=> Aggregate = aggregate;
}
/// <summary>
/// One row returned by <see cref="IHistorianDataSource.ReadEventsAsync"/> — a historical
/// alarm/event record, not the OPC UA live-event stream. Fields match the minimum set the
/// Server needs to populate a <c>HistoryEventFieldList</c> for HistoryReadEvents responses.
/// </summary>
/// <param name="EventId">Stable unique id for the event — driver-specific format.</param>
/// <param name="SourceName">Source object that emitted the event. May differ from the <c>sourceName</c> filter the caller passed (fuzzy matches).</param>
/// <param name="EventTimeUtc">Process-side timestamp — when the event actually occurred.</param>
/// <param name="ReceivedTimeUtc">Historian-side timestamp — when the historian persisted the row; may lag <paramref name="EventTimeUtc"/> by the historian's buffer flush cadence.</param>
/// <param name="Message">Human-readable message text.</param>
/// <param name="Severity">OPC UA severity (1-1000). Backends map their native priority scale onto this range.</param>
public sealed record HistoricalEvent(
string EventId,
string? SourceName,
DateTime EventTimeUtc,
DateTime ReceivedTimeUtc,
string? Message,
ushort Severity);
/// <summary>Result of a <see cref="IHistorianDataSource.ReadEventsAsync"/> call.</summary>
/// <param name="Events">Events in chronological order by <c>EventTimeUtc</c>.</param>
/// <param name="ContinuationPoint">Opaque token for the next call when more events are available; null when complete.</param>
public sealed record HistoricalEventsResult(
IReadOnlyList<HistoricalEvent> Events,
byte[]? ContinuationPoint);
src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorianDataSource.cs
+4 -12
View File
@@ -5,15 +5,9 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
/// and resolved per OPC UA namespace, independent of any driver's lifecycle.
/// </summary>
/// <remarks>
/// Distinct from <see cref="IHistoryProvider"/>:
/// <list type="bullet">
/// <item><see cref="IHistoryProvider"/> is a *driver capability* — the server
/// dispatches to it via the driver instance.</item>
/// <item><see cref="IHistorianDataSource"/> is a *server registration* — the
/// server resolves it via namespace and calls it directly, so a single
/// historian (e.g. Wonderware) can serve many drivers' nodes, and drivers can
/// restart without dropping history availability.</item>
/// </list>
/// This is a *server registration*, not a driver capability: drivers carry no history
/// surface (Gitea #655). The server calls it directly, so a single historian can serve
/// many drivers' nodes, and drivers can restart without dropping history availability.
/// All values returned use the shared <see cref="DataValueSnapshot"/> /
/// <see cref="HistoricalEvent"/> shapes; backend-specific quality / type encodings
/// are translated to OPC UA <c>StatusCode</c> uints inside the data source.
@@ -86,9 +80,7 @@ public interface IHistorianDataSource : IDisposable
/// equivalent "use default" sentinel.
///
/// This surface declares <see cref="ReadAtTimeAsync"/> and <see cref="ReadEventsAsync"/>
/// as required members — a server-side historian owns the full read surface, unlike
/// <see cref="IHistoryProvider"/> where the same two methods are optional default-impl
/// methods so legacy drivers can stay raw-only. The asymmetry is intentional.
/// as required members — a server-side historian owns the full read surface.
/// </remarks>
/// <param name="sourceName">The source name to filter events, or null to return events from all sources.</param>
/// <param name="startUtc">The start of the time range in UTC.</param>
src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IDriver.cs
+1 -1
View File
@@ -4,7 +4,7 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
/// Required capability for every driver instance. Owns lifecycle, metadata, health.
/// Other capabilities (<see cref="ITagDiscovery"/>, <see cref="IReadable"/>,
/// <see cref="IWritable"/>, <see cref="ISubscribable"/>, <see cref="IAlarmSource"/>,
/// <see cref="IHistoryProvider"/>, <see cref="IRediscoverable"/>,
/// <see cref="IRediscoverable"/>,
/// <see cref="IHostConnectivityProbe"/>) are composable — a driver implements only what its
/// backend actually supports.
/// </summary>
src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/IHistoryProvider.cs
-237
View File
@@ -1,237 +0,0 @@
namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
/// <summary>
/// Driver capability for historical-data reads (OPC UA HistoryRead). Optional —
/// only drivers whose backends carry historian data implement this. Currently:
/// Galaxy (Wonderware Historian via the optional plugin), OPC UA Client (forward
/// to upstream server).
/// </summary>
/// <remarks>
/// <see cref="ReadAtTimeAsync"/> and <see cref="ReadEventsAsync"/> are C# default interface
/// methods that throw <see cref="NotSupportedException"/> — drivers opt in by overriding so
/// a raw-only driver compiles without forcing it to provide at-time / event surfaces it
/// has no backend for. The sibling server-side surface, <see cref="IHistorianDataSource"/>,
/// declares both methods as required because a registered historian owns the full read
/// surface; the asymmetry is intentional.
/// </remarks>
public interface IHistoryProvider
{
/// <summary>
/// Read raw historical samples for a single attribute over a time range.
/// The Core wraps this with continuation-point handling.
/// </summary>
/// <param name="fullReference">The full reference path to the attribute.</param>
/// <param name="startUtc">Inclusive lower bound on the time range (UTC).</param>
/// <param name="endUtc">Exclusive upper bound on the time range (UTC).</param>
/// <param name="maxValuesPerNode">Maximum number of values to return per node.</param>
/// <param name="cancellationToken">A cancellation token to stop the operation.</param>
/// <returns>A task that returns the history read result containing samples and optional continuation point.</returns>
Task<HistoryReadResult> ReadRawAsync(
string fullReference,
DateTime startUtc,
DateTime endUtc,
uint maxValuesPerNode,
CancellationToken cancellationToken);
/// <summary>
/// Read processed (aggregated) samples — interval-bucketed average / min / max / etc.
/// Optional — drivers that only support raw history can throw <see cref="NotSupportedException"/>.
/// </summary>
/// <param name="fullReference">The full reference path to the attribute.</param>
/// <param name="startUtc">Inclusive lower bound on the time range (UTC).</param>
/// <param name="endUtc">Exclusive upper bound on the time range (UTC).</param>
/// <param name="interval">The time interval for bucketing samples.</param>
/// <param name="aggregate">The aggregate function to apply to each bucket.</param>
/// <param name="cancellationToken">A cancellation token to stop the operation.</param>
/// <returns>A task that returns the history read result containing aggregated samples and optional continuation point.</returns>
Task<HistoryReadResult> ReadProcessedAsync(
string fullReference,
DateTime startUtc,
DateTime endUtc,
TimeSpan interval,
HistoryAggregateType aggregate,
CancellationToken cancellationToken);
/// <summary>
/// Read one sample per requested timestamp — OPC UA HistoryReadAtTime service. The
/// driver interpolates (or returns the prior-boundary sample) when no exact match
/// exists. Optional; drivers that can't interpolate throw <see cref="NotSupportedException"/>.
/// </summary>
/// <remarks>
/// Default implementation throws. Drivers opt in by overriding; keeps existing
/// <c>IHistoryProvider</c> implementations compiling without forcing a ReadAtTime path
/// they may not have a backend for.
/// </remarks>
/// <param name="fullReference">The full reference path to the attribute.</param>
/// <param name="timestampsUtc">The list of timestamps at which to read values.</param>
/// <param name="cancellationToken">A cancellation token to stop the operation.</param>
/// <returns>A task that returns the history read result containing samples at the requested times.</returns>
Task<HistoryReadResult> ReadAtTimeAsync(
string fullReference,
IReadOnlyList<DateTime> timestampsUtc,
CancellationToken cancellationToken)
=> throw new NotSupportedException(
$"{GetType().Name} does not implement ReadAtTimeAsync. " +
"Drivers whose backends support at-time reads override this method.");
/// <summary>
/// Read historical alarm/event records — OPC UA HistoryReadEvents service. Distinct
/// from the live event stream — historical rows come from an event historian (Galaxy's
/// Alarm Provider history log, etc.) rather than the driver's active subscription.
/// </summary>
/// <param name="sourceName">
/// Optional filter: null means "all sources", otherwise restrict to events from that
/// source-object name. Drivers may ignore the filter if the backend doesn't support it.
/// </param>
/// <param name="startUtc">Inclusive lower bound on <c>EventTimeUtc</c>.</param>
/// <param name="endUtc">Exclusive upper bound on <c>EventTimeUtc</c>.</param>
/// <param name="maxEvents">
/// Upper cap on returned events — the driver's backend enforces this. The type is
/// <see cref="int"/> rather than <see cref="uint"/> (which the sibling raw / processed
/// reads use for <c>maxValuesPerNode</c>) because callers and downstream historian
/// adapters historically treat <c>maxEvents &lt;= 0</c> as a sentinel meaning
/// "use the backend's default cap" (see <c>WonderwareHistorianClient</c> /
/// <c>HistorianDataSource</c>). The asymmetry is intentional.
///
/// <b>Continuation contract when using the sentinel:</b> When <c>maxEvents &lt;= 0</c>
/// the backend applies its own cap. If the backend's cap truncates the result
/// (i.e. more matching events exist beyond the returned set), implementations MUST set
/// <see cref="HistoricalEventsResult.ContinuationPoint"/> to a non-null token so
/// callers can detect truncation and page. A null <c>ContinuationPoint</c> means all
/// matching events were returned — callers rely on this to decide whether to page.
/// </param>
/// <param name="cancellationToken">Request cancellation.</param>
/// <returns>A task that returns the historical events result containing matching event records and optional continuation point.</returns>
/// <remarks>
/// Default implementation throws. Drivers whose backend can serve historical events
/// override: Galaxy (Wonderware Alarm &amp; Events log) and the OPC UA Client driver
/// (forwarding HistoryReadEvents to its upstream server). Raw-only drivers such as Modbus
/// stay with the default and let callers see <c>BadHistoryOperationUnsupported</c>.
///
/// Note the type asymmetry with <see cref="ReadRawAsync"/> /
/// <see cref="ReadProcessedAsync"/> (both use <c>uint maxValuesPerNode</c>): event
/// readers accept a signed <c>int maxEvents</c> so callers can pass 0 / negative as a
/// "use default cap" sentinel without an extra parameter or overload.
/// </remarks>
Task<HistoricalEventsResult> ReadEventsAsync(
string? sourceName,
DateTime startUtc,
DateTime endUtc,
int maxEvents,
CancellationToken cancellationToken)
=> throw new NotSupportedException(
$"{GetType().Name} does not implement ReadEventsAsync. " +
"Drivers whose backends have an event historian override this method.");
}
/// <summary>Result of a HistoryRead call.</summary>
/// <param name="Samples">Returned samples in chronological order.</param>
/// <param name="ContinuationPoint">Opaque token for the next call when more samples are available; null when complete.</param>
public sealed record HistoryReadResult(
IReadOnlyList<DataValueSnapshot> Samples,
byte[]? ContinuationPoint);
/// <summary>Aggregate function for processed history reads. Mirrors OPC UA Part 13 standard aggregates.</summary>
/// <remarks>
/// The first five members are the original set every historian backend serves. The remainder (#285) are
/// the further standard Part 13 aggregates a server-side HistoryRead can forward to a backend that
/// supports them (the OPC UA Client driver forwards all of them upstream). A backend that cannot serve a
/// member must throw <see cref="HistoryAggregateNotSupportedException"/> — the server maps that to
/// <c>BadAggregateNotSupported</c> rather than a generic failure. <b>Append only</b>: the numeric values
/// are not persisted, but every <c>switch</c> over this enum must keep a total default.
/// </remarks>
public enum HistoryAggregateType
{
Average,
Minimum,
Maximum,
Total,
Count,
Interpolative,
TimeAverage,
Start,
End,
Delta,
Range,
MinimumActualTime,
MaximumActualTime,
StartBound,
EndBound,
DeltaBounds,
DurationGood,
DurationBad,
PercentGood,
PercentBad,
WorstQuality,
NumberOfTransitions,
AnnotationCount,
StandardDeviationPopulation,
VariancePopulation,
}
/// <summary>
/// Thrown by a history backend when it cannot serve the requested <see cref="HistoryAggregateType"/>.
/// The Server HistoryRead path translates it to <c>BadAggregateNotSupported</c> for that node only.
/// </summary>
public sealed class HistoryAggregateNotSupportedException : NotSupportedException
{
/// <summary>The aggregate the backend could not serve.</summary>
public HistoryAggregateType Aggregate { get; }
/// <summary>Creates the exception for <paramref name="aggregate"/>.</summary>
/// <param name="aggregate">The unsupported aggregate.</param>
public HistoryAggregateNotSupportedException(HistoryAggregateType aggregate)
: base($"The historian backend does not support the {aggregate} aggregate.")
=> Aggregate = aggregate;
}
/// <summary>
/// Thrown by a history backend that is a passthrough to another OPC UA server when that server answered
/// a node with a Bad per-node <c>HistoryReadResult.StatusCode</c> other than
/// <c>BadAggregateNotSupported</c> (which has its own <see cref="HistoryAggregateNotSupportedException"/>).
/// The Server HistoryRead path surfaces <see cref="StatusCode"/> verbatim as that node's result status
/// instead of the generic <c>BadHistoryOperationUnsupported</c>, so a refused read is never mistaken for
/// an empty Good one.
/// </summary>
public sealed class HistoryReadStatusException : Exception
{
/// <summary>The raw OPC UA status code the upstream server returned for the node.</summary>
public uint StatusCode { get; }
/// <summary>Creates the exception for the upstream <paramref name="statusCode"/>.</summary>
/// <param name="statusCode">The Bad OPC UA status code to surface for the node.</param>
/// <exception cref="ArgumentException"><paramref name="statusCode"/> is not a Bad code (severity bits <c>10</c>/<c>11</c>).</exception>
public HistoryReadStatusException(uint statusCode)
: base(IsBad(statusCode)
? $"The upstream server refused the history read with status 0x{statusCode:X8}."
: throw new ArgumentException($"0x{statusCode:X8} is not a Bad status code.", nameof(statusCode)))
=> StatusCode = statusCode;
private static bool IsBad(uint statusCode) => (statusCode & 0x80000000u) != 0;
}
/// <summary>
/// One row returned by <see cref="IHistoryProvider.ReadEventsAsync"/> — a historical
/// alarm/event record, not the OPC UA live-event stream. Fields match the minimum set the
/// Server needs to populate a <c>HistoryEventFieldList</c> for HistoryReadEvents responses.
/// </summary>
/// <param name="EventId">Stable unique id for the event — driver-specific format.</param>
/// <param name="SourceName">Source object that emitted the event. May differ from the <c>sourceName</c> filter the caller passed (fuzzy matches).</param>
/// <param name="EventTimeUtc">Process-side timestamp — when the event actually occurred.</param>
/// <param name="ReceivedTimeUtc">Historian-side timestamp — when the historian persisted the row; may lag <paramref name="EventTimeUtc"/> by the historian's buffer flush cadence.</param>
/// <param name="Message">Human-readable message text.</param>
/// <param name="Severity">OPC UA severity (1-1000). Drivers map their native priority scale onto this range.</param>
public sealed record HistoricalEvent(
string EventId,
string? SourceName,
DateTime EventTimeUtc,
DateTime ReceivedTimeUtc,
string? Message,
ushort Severity);
/// <summary>Result of a <see cref="IHistoryProvider.ReadEventsAsync"/> call.</summary>
/// <param name="Events">Events in chronological order by <c>EventTimeUtc</c>.</param>
/// <param name="ContinuationPoint">Opaque token for the next call when more events are available; null when complete.</param>
public sealed record HistoricalEventsResult(
IReadOnlyList<HistoricalEvent> Events,
byte[]? ContinuationPoint);
src/Core/ZB.MOM.WW.OtOpcUa.Core/Resilience/DriverResilienceOptions.cs
-3
View File
@@ -81,7 +81,6 @@ public sealed record DriverResilienceOptions
[DriverCapability.Probe] = new(TimeoutSeconds: 2, RetryCount: 3, BreakerFailureThreshold: 5),
[DriverCapability.AlarmSubscribe] = new(TimeoutSeconds: 5, RetryCount: 3, BreakerFailureThreshold: 5),
[DriverCapability.AlarmAcknowledge] = new(TimeoutSeconds: 5, RetryCount: 0, BreakerFailureThreshold: 5),
[DriverCapability.HistoryRead] = new(TimeoutSeconds: 30, RetryCount: 2, BreakerFailureThreshold: 5),
// Initialize: no retry (the DriverInstanceActor reconnect timer is the retry loop) and a
// threshold of 2, NOT the Read-sized 5 — the breaker samples over 30 s and the reconnect
// timer fires every 10 s, so at most three init attempts fit one window; a threshold of 5
@@ -97,7 +96,6 @@ public sealed record DriverResilienceOptions
[DriverCapability.Probe] = new(TimeoutSeconds: 4, RetryCount: 3, BreakerFailureThreshold: 5),
[DriverCapability.AlarmSubscribe] = new(TimeoutSeconds: 8, RetryCount: 3, BreakerFailureThreshold: 5),
[DriverCapability.AlarmAcknowledge] = new(TimeoutSeconds: 8, RetryCount: 0, BreakerFailureThreshold: 5),
[DriverCapability.HistoryRead] = new(TimeoutSeconds: 60, RetryCount: 2, BreakerFailureThreshold: 5),
// See the Tier A Initialize note: threshold 2 is cadence-reachability, not a tuning choice.
[DriverCapability.Initialize] = new(TimeoutSeconds: 8, RetryCount: 0, BreakerFailureThreshold: 2),
},
@@ -110,7 +108,6 @@ public sealed record DriverResilienceOptions
[DriverCapability.Probe] = new(TimeoutSeconds: 10, RetryCount: 1, BreakerFailureThreshold: 0),
[DriverCapability.AlarmSubscribe] = new(TimeoutSeconds: 15, RetryCount: 1, BreakerFailureThreshold: 0),
[DriverCapability.AlarmAcknowledge] = new(TimeoutSeconds: 15, RetryCount: 0, BreakerFailureThreshold: 0),
[DriverCapability.HistoryRead] = new(TimeoutSeconds: 120, RetryCount: 1, BreakerFailureThreshold: 0),
// Tier C stays breaker-less like every other Tier C capability (process-level supervision).
[DriverCapability.Initialize] = new(TimeoutSeconds: 15, RetryCount: 0, BreakerFailureThreshold: 0),
},
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs
+1 -330
View File
@@ -29,7 +29,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient;
/// monitored-item handles. That mechanic lands in PR 69.
/// </para>
/// </remarks>
public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IAlarmSource, IHistoryProvider, ISkippedTagReporter, IDisposable, IAsyncDisposable
public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IAlarmSource, ISkippedTagReporter, IDisposable, IAsyncDisposable
{
/// <summary>Authored tags whose TagConfig carried no usable <c>nodeId</c>, so there is no upstream node
/// to dial (Gitea #541). This driver's skip path was the quietest of the twelve: <c>NamespaceMap</c>
@@ -2258,335 +2258,6 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
public string DiagnosticId => $"opcua-alarm-sub-{Id}";
}
// ---- IHistoryProvider (passthrough to upstream server) ----
/// <inheritdoc />
public async Task<Core.Abstractions.HistoryReadResult> ReadRawAsync(
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
CancellationToken cancellationToken)
{
var details = new ReadRawModifiedDetails
{
IsReadModified = false,
StartTime = startUtc,
EndTime = endUtc,
NumValuesPerNode = maxValuesPerNode,
ReturnBounds = false,
};
return await ExecuteHistoryReadAsync(fullReference, new ExtensionObject(details), cancellationToken)
.ConfigureAwait(false);
}
/// <inheritdoc />
public async Task<Core.Abstractions.HistoryReadResult> ReadProcessedAsync(
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
HistoryAggregateType aggregate, CancellationToken cancellationToken)
{
var aggregateId = MapAggregateToNodeId(aggregate);
var details = new ReadProcessedDetails
{
StartTime = startUtc,
EndTime = endUtc,
ProcessingInterval = interval.TotalMilliseconds,
AggregateType = [aggregateId],
};
return await ExecuteHistoryReadAsync(fullReference, new ExtensionObject(details), cancellationToken, aggregate)
.ConfigureAwait(false);
}
/// <inheritdoc />
public async Task<Core.Abstractions.HistoryReadResult> ReadAtTimeAsync(
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken)
{
var reqTimes = new DateTimeCollection(timestampsUtc);
var details = new ReadAtTimeDetails
{
ReqTimes = reqTimes,
UseSimpleBounds = true,
};
return await ExecuteHistoryReadAsync(fullReference, new ExtensionObject(details), cancellationToken)
.ConfigureAwait(false);
}
/// <summary>
/// Shared HistoryRead wire path — used by Raw/Processed/AtTime. Handles NodeId parse,
/// Session.HistoryReadAsync call, per-sample StatusCode passthrough (no translation per §8
/// cascading-quality rule), and HistoryData unwrap into <see cref="DataValueSnapshot"/>. A Bad
/// per-NODE <c>HistoryReadResult.StatusCode</c> is never read as empty data: <c>BadAggregateNotSupported</c>
/// throws <see cref="HistoryAggregateNotSupportedException"/>, any other Bad throws
/// <see cref="HistoryReadStatusException"/> (Gitea #649). Good / GoodNoData / GoodMoreData / Uncertain are
/// unchanged.
/// </summary>
private async Task<Core.Abstractions.HistoryReadResult> ExecuteHistoryReadAsync(
string fullReference, ExtensionObject historyReadDetails, CancellationToken ct,
HistoryAggregateType? aggregate = null)
{
// Make sure a session exists before queuing on the gate, but do NOT bind the wire call
// (or the namespace-relative NodeId parse) to this reference — a reconnect can swap
// Session while we wait on _gate. The session actually used is re-read inside the gate.
_ = RequireSession();
await _gate.WaitAsync(ct).ConfigureAwait(false);
try
{
// Re-read Session inside the critical section: if a reconnect completed while we
// were blocked on _gate, OnReconnectComplete has already swapped in the new session.
// NodeId parsing is namespace-relative, so it must also use the current session's
// namespace table.
var session = Session;
if (session is null || !TryResolveTagNode(session, fullReference, out var nodeId))
{
return new Core.Abstractions.HistoryReadResult([], null);
}
var nodesToRead = new HistoryReadValueIdCollection
{
new HistoryReadValueId { NodeId = nodeId },
};
var resp = await session.HistoryReadAsync(
requestHeader: null,
historyReadDetails: historyReadDetails,
timestampsToReturn: TimestampsToReturn.Both,
releaseContinuationPoints: false,
nodesToRead: nodesToRead,
ct: ct).ConfigureAwait(false);
if (resp.Results.Count == 0) return new Core.Abstractions.HistoryReadResult([], null);
var r = resp.Results[0];
// The per-node status decides whether the payload means anything. BadNoData is the one Bad code
// that legitimately means "nothing in the window", so it stays an empty result.
// CodeBits (not Code) so an info/overflow sub-status bit on the upstream code does not dodge the match.
if (StatusCode.IsBad(r.StatusCode) && r.StatusCode.CodeBits != StatusCodes.BadNoData)
{
if (r.StatusCode.CodeBits == StatusCodes.BadAggregateNotSupported && aggregate is { } agg)
throw new HistoryAggregateNotSupportedException(agg);
throw new HistoryReadStatusException(r.StatusCode.Code);
}
// Unwrap HistoryData from the ExtensionObject-encoded payload the SDK returns.
// Samples stay in chronological order per OPC UA Part 11; cascading-quality
// rule: preserve each DataValue's upstream StatusCode + timestamps verbatim.
var samples = new List<DataValueSnapshot>();
if (r.HistoryData?.Body is HistoryData hd)
{
var now = DateTime.UtcNow;
foreach (var dv in hd.DataValues)
{
samples.Add(new DataValueSnapshot(
Value: dv.Value,
StatusCode: dv.StatusCode.Code,
SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp,
ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? now : dv.ServerTimestamp));
}
}
var contPt = r.ContinuationPoint is { Length: > 0 } ? r.ContinuationPoint : null;
return new Core.Abstractions.HistoryReadResult(samples, contPt);
}
finally { _gate.Release(); }
}
/// <summary>Map <see cref="HistoryAggregateType"/> to the OPC UA Part 13 standard aggregate NodeId.</summary>
/// <param name="aggregate">The aggregation function type to map.</param>
/// <returns>The standard aggregate function NodeId.</returns>
internal static NodeId MapAggregateToNodeId(HistoryAggregateType aggregate) => aggregate switch
{
HistoryAggregateType.Average => ObjectIds.AggregateFunction_Average,
HistoryAggregateType.Minimum => ObjectIds.AggregateFunction_Minimum,
HistoryAggregateType.Maximum => ObjectIds.AggregateFunction_Maximum,
HistoryAggregateType.Total => ObjectIds.AggregateFunction_Total,
HistoryAggregateType.Count => ObjectIds.AggregateFunction_Count,
HistoryAggregateType.Interpolative => ObjectIds.AggregateFunction_Interpolative,
HistoryAggregateType.TimeAverage => ObjectIds.AggregateFunction_TimeAverage,
HistoryAggregateType.Start => ObjectIds.AggregateFunction_Start,
HistoryAggregateType.End => ObjectIds.AggregateFunction_End,
HistoryAggregateType.Delta => ObjectIds.AggregateFunction_Delta,
HistoryAggregateType.Range => ObjectIds.AggregateFunction_Range,
HistoryAggregateType.MinimumActualTime => ObjectIds.AggregateFunction_MinimumActualTime,
HistoryAggregateType.MaximumActualTime => ObjectIds.AggregateFunction_MaximumActualTime,
HistoryAggregateType.StartBound => ObjectIds.AggregateFunction_StartBound,
HistoryAggregateType.EndBound => ObjectIds.AggregateFunction_EndBound,
HistoryAggregateType.DeltaBounds => ObjectIds.AggregateFunction_DeltaBounds,
HistoryAggregateType.DurationGood => ObjectIds.AggregateFunction_DurationGood,
HistoryAggregateType.DurationBad => ObjectIds.AggregateFunction_DurationBad,
HistoryAggregateType.PercentGood => ObjectIds.AggregateFunction_PercentGood,
HistoryAggregateType.PercentBad => ObjectIds.AggregateFunction_PercentBad,
HistoryAggregateType.WorstQuality => ObjectIds.AggregateFunction_WorstQuality,
HistoryAggregateType.NumberOfTransitions => ObjectIds.AggregateFunction_NumberOfTransitions,
HistoryAggregateType.AnnotationCount => ObjectIds.AggregateFunction_AnnotationCount,
HistoryAggregateType.StandardDeviationPopulation => ObjectIds.AggregateFunction_StandardDeviationPopulation,
HistoryAggregateType.VariancePopulation => ObjectIds.AggregateFunction_VariancePopulation,
_ => throw new ArgumentOutOfRangeException(nameof(aggregate), aggregate, null),
};
// Canonical BaseEventType select-clause order — MapHistoryEvents maps by these indices.
private static readonly string[] EventFieldBrowseNames =
[
BrowseNames.EventId, // 0
BrowseNames.SourceName, // 1
BrowseNames.Time, // 2
BrowseNames.ReceiveTime, // 3
BrowseNames.Message, // 4
BrowseNames.Severity, // 5
];
/// <summary>
/// Builds the fixed canonical EventFilter the driver sends upstream for HistoryReadEvents —
/// the six BaseEventType fields the OtOpcUa server projects (<see cref="HistoricalEvent"/>).
/// The clause order is load-bearing: <see cref="MapHistoryEvents"/> reads results by index.
/// </summary>
/// <returns>An EventFilter with six SimpleAttributeOperand value clauses.</returns>
internal static EventFilter BuildBaseEventFilter()
{
var filter = new EventFilter();
foreach (var browseName in EventFieldBrowseNames)
{
filter.SelectClauses.Add(new SimpleAttributeOperand
{
TypeDefinitionId = ObjectTypeIds.BaseEventType,
BrowsePath = new QualifiedNameCollection { new QualifiedName(browseName) },
AttributeId = Attributes.Value,
});
}
return filter;
}
/// <summary>
/// Maps an upstream <see cref="HistoryEvent"/> (field arrays ordered to match
/// <see cref="BuildBaseEventFilter"/>) onto <see cref="HistoricalEvent"/> records. Defensive:
/// short / null / wrong-typed fields degrade to null/default rather than throwing.
/// </summary>
/// <param name="historyEvent">The upstream history-event payload.</param>
/// <returns>The mapped historical events in upstream order.</returns>
internal static IReadOnlyList<Core.Abstractions.HistoricalEvent> MapHistoryEvents(HistoryEvent historyEvent)
{
if (historyEvent?.Events is not { Count: > 0 } rows) return [];
var result = new List<Core.Abstractions.HistoricalEvent>(rows.Count);
foreach (var row in rows)
{
var fields = row?.EventFields;
result.Add(new Core.Abstractions.HistoricalEvent(
EventId: CoerceEventId(FieldAt(fields, 0)),
SourceName: CoerceString(FieldAt(fields, 1)),
EventTimeUtc: CoerceDateTime(FieldAt(fields, 2)),
ReceivedTimeUtc: CoerceDateTime(FieldAt(fields, 3)),
Message: CoerceString(FieldAt(fields, 4)),
Severity: CoerceSeverity(FieldAt(fields, 5))));
}
return result;
}
private static object? FieldAt(VariantCollection? fields, int index)
=> fields is not null && index < fields.Count ? fields[index].Value : null;
private static string CoerceEventId(object? value) => value switch
{
byte[] bytes => Convert.ToBase64String(bytes),
string s => s,
null => string.Empty,
_ => value.ToString() ?? string.Empty,
};
private static string? CoerceString(object? value) => value switch
{
LocalizedText lt => lt.Text,
string s => s,
null => null,
_ => value.ToString(),
};
// Missing-field sentinel is UTC-kinded so a downstream .ToUniversalTime() can't shift it.
private static readonly DateTime MissingTimeSentinel =
DateTime.SpecifyKind(DateTime.MinValue, DateTimeKind.Utc);
private static DateTime CoerceDateTime(object? value)
=> value is DateTime dt ? dt : MissingTimeSentinel;
private static ushort CoerceSeverity(object? value)
{
try { return value is null ? (ushort)0 : Convert.ToUInt16(value); }
catch (Exception ex) when (ex is FormatException or InvalidCastException or OverflowException) { return 0; }
}
/// <inheritdoc />
// Forwards OPC UA HistoryReadEvents to the upstream server. Sends the fixed canonical
// BuildBaseEventFilter and maps the result onto HistoricalEvent (the OtOpcUa server projects
// only those six BaseEventType fields, so a richer client filter would be discarded
// server-side — the driver supplies the canonical set itself). sourceName is the upstream
// event-notifier NodeId to read from (mirrors how fullReference is the upstream NodeId for
// raw reads); null/empty resolves to the upstream Server object (i=2253), the standard
// server-wide event notifier, and an unparseable id short-circuits to an empty result,
// matching the raw path's malformed-NodeId behavior.
public async Task<Core.Abstractions.HistoricalEventsResult> ReadEventsAsync(
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
CancellationToken cancellationToken)
{
// Confirm a session exists before queuing; the session actually used — and the
// namespace-relative source-NodeId parse — are re-read inside the gate so a reconnect
// mid-wait can't leave us reading from a closed session.
_ = RequireSession();
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
// Re-read Session inside the critical section: OnReconnectComplete may have swapped
// in a new session while we were blocked on _gate, and the source-NodeId parse is
// namespace-relative so it must bind against the current session's namespace table.
var session = Session;
if (session is null) return new Core.Abstractions.HistoricalEventsResult([], null);
NodeId notifierNodeId;
if (string.IsNullOrEmpty(sourceName))
{
notifierNodeId = ObjectIds.Server;
}
// sourceName is an upstream event-notifier node id (not a tag RawPath), so it
// re-binds directly against the session via the static resolve.
else if (!NamespaceMap.TryResolve(session, sourceName, out var parsed))
{
return new Core.Abstractions.HistoricalEventsResult([], null);
}
else
{
notifierNodeId = parsed;
}
var details = new ReadEventDetails
{
StartTime = startUtc,
EndTime = endUtc,
NumValuesPerNode = maxEvents <= 0 ? 0u : (uint)maxEvents,
Filter = BuildBaseEventFilter(),
};
var nodesToRead = new HistoryReadValueIdCollection
{
new HistoryReadValueId { NodeId = notifierNodeId },
};
var resp = await session.HistoryReadAsync(
requestHeader: null,
historyReadDetails: new ExtensionObject(details),
timestampsToReturn: TimestampsToReturn.Both,
releaseContinuationPoints: false,
nodesToRead: nodesToRead,
ct: cancellationToken).ConfigureAwait(false);
if (resp.Results.Count == 0) return new Core.Abstractions.HistoricalEventsResult([], null);
var r = resp.Results[0];
var events = r.HistoryData?.Body is HistoryEvent he
? MapHistoryEvents(he)
: [];
var contPt = r.ContinuationPoint is { Length: > 0 } ? r.ContinuationPoint : null;
return new Core.Abstractions.HistoricalEventsResult(events, contPt);
}
finally { _gate.Release(); }
}
// ---- IHostConnectivityProbe ----
/// <summary>
src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/ResilienceFormModel.cs
+1 -1
View File
@@ -21,7 +21,7 @@ namespace ZB.MOM.WW.OtOpcUa.AdminUI.Components.Shared.Drivers;
public sealed class ResilienceFormModel
{
public static readonly string[] Capabilities =
["Read", "Write", "Discover", "Subscribe", "Probe", "AlarmSubscribe", "AlarmAcknowledge", "HistoryRead", "Initialize"];
["Read", "Write", "Discover", "Subscribe", "Probe", "AlarmSubscribe", "AlarmAcknowledge", "Initialize"];
// RecycleIntervalSeconds is GONE (Gitea #522) — the scheduled-recycle machinery it authored was
// deleted, so the field configured nothing. It is deliberately NOT explicitly stripped from stored
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs
+1 -1
View File
@@ -193,7 +193,7 @@ internal static class HistoryPaging
/// <param name="boundaryT">The single SourceTimestamp every tie in the cluster shares.</param>
/// <param name="endUtc">The (inclusive) upper bound of the read window; unchanged across pages.
/// OpcUaServer-003, accepted by design: the backend contract
/// (<c>IHistoryProvider.ReadRawAsync</c>) treats <c>endUtc</c> as EXCLUSIVE, so the
/// (<c>IHistorianDataSource.ReadRawAsync</c>) treats <c>endUtc</c> as EXCLUSIVE, so the
/// <c>next &lt;= endUtc</c> test here is deliberately the conservative side — at worst one extra
/// empty resume page (<c>[endUtc, endUtc)</c> → GoodNoData, no continuation point), never a
/// dropped or duplicated sample. Tightening to <c>&lt;</c> would be correct only if the live
src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs
-15
View File
@@ -5419,14 +5419,6 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
errors[handle.Index] = StatusCodes.BadAggregateNotSupported;
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadAggregateNotSupported };
}
catch (HistoryReadStatusException ex)
{
// A passthrough backend (OpcUaClient) relayed the upstream server's Bad per-node status - surface
// it verbatim for THIS node rather than collapsing it to BadHistoryOperationUnsupported.
_logger.LogDebug(ex, "OtOpcUaNodeManager: backend refused history read for node {NodeId}", handle.NodeId);
errors[handle.Index] = new StatusCode(ex.StatusCode);
results[handle.Index] = new SdkHistoryReadResult { StatusCode = new StatusCode(ex.StatusCode) };
}
catch (Exception ex)
{
// One node's backend failure (throw / timeout) must not throw out of the batch — surface a
@@ -5628,13 +5620,6 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
errors[handle.Index] = StatusCodes.BadTimeout;
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTimeout };
}
catch (HistoryReadStatusException ex)
{
// Upstream Bad per-node status relayed by a passthrough backend - surface it verbatim (see ServeNodeAsync).
_logger.LogDebug(ex, "OtOpcUaNodeManager: backend refused HistoryReadRaw for node {NodeId}", handle.NodeId);
errors[handle.Index] = new StatusCode(ex.StatusCode);
results[handle.Index] = new SdkHistoryReadResult { StatusCode = new StatusCode(ex.StatusCode) };
}
catch (Exception ex)
{
// One node's backend failure must not throw out of the batch — Bad for THIS node only.
src/Tooling/ZB.MOM.WW.OtOpcUa.Analyzers/UnwrappedCapabilityCallAnalyzer.cs
+3 -4
View File
@@ -26,8 +26,8 @@ namespace ZB.MOM.WW.OtOpcUa.Analyzers;
/// <c>IReadable.ReadAsync</c> still trips the rule.
/// </para>
/// <para>
/// <b>Default-interface-method handling:</b> <c>IHistoryProvider.ReadAtTimeAsync</c> and
/// <c>IHistoryProvider.ReadEventsAsync</c> ship as DIM bodies. When a driver inherits the
/// <b>Default-interface-method handling:</b> a guarded interface member may ship as a DIM
/// body. When a driver inherits the
/// DIM (no override), <c>FindImplementationForInterfaceMember</c> returns the interface's
/// own method symbol, which still equals <paramref name="method"/> for an interface-typed
/// receiver. When a driver overrides the DIM, the override symbol equals
@@ -75,7 +75,6 @@ public sealed class UnwrappedCapabilityCallAnalyzer : DiagnosticAnalyzer
"ZB.MOM.WW.OtOpcUa.Core.Abstractions.ISubscribable",
"ZB.MOM.WW.OtOpcUa.Core.Abstractions.IHostConnectivityProbe",
"ZB.MOM.WW.OtOpcUa.Core.Abstractions.IAlarmSource",
"ZB.MOM.WW.OtOpcUa.Core.Abstractions.IHistoryProvider",
];
/// <summary>
@@ -106,7 +105,7 @@ public sealed class UnwrappedCapabilityCallAnalyzer : DiagnosticAnalyzer
category: "OtOpcUa.Resilience",
defaultSeverity: DiagnosticSeverity.Warning,
isEnabledByDefault: true,
description: "Phase 6.1 Stream A requires every IReadable/IWritable/ITagDiscovery/ISubscribable/IHostConnectivityProbe/IAlarmSource/IHistoryProvider call to route through the shared Polly pipeline. Direct calls skip the pipeline + lose per-host isolation, retry semantics, and telemetry. If the caller is Core/Server/Driver dispatch code, wrap the call in CapabilityInvoker.ExecuteAsync. If the caller is a unit test invoking the driver directly to test its wire-level behavior, either suppress with a pragma or move the suppression into a NoWarn for the test project.");
description: "Phase 6.1 Stream A requires every IReadable/IWritable/ITagDiscovery/ISubscribable/IHostConnectivityProbe/IAlarmSource call to route through the shared Polly pipeline. Direct calls skip the pipeline + lose per-host isolation, retry semantics, and telemetry. If the caller is Core/Server/Driver dispatch code, wrap the call in CapabilityInvoker.ExecuteAsync. If the caller is a unit test invoking the driver directly to test its wire-level behavior, either suppress with a pragma or move the suppression into a NoWarn for the test project.");
/// <inheritdoc />
public override ImmutableArray<DiagnosticDescriptor> SupportedDiagnostics { get; } = ImmutableArray.Create(Rule);