fix(historian): #674 review follow-ups — HistoryUpdate boot-window gate, node-level v1 refusal, plan-before-gate, UTC skew compare
- HistoryUpdate Primary gate now reads IHistoryUpdateRoleView (PrimaryGatePolicy.ShouldServeHistoryUpdate): the data-plane verdict but CLOSED until this node's first redundancy snapshot, so both pair nodes cannot accept in the pre-join boot window. Single nodes open on their own singleton's first snapshot; admin-only nodes register no view => no gate. DriverHostActor publishes it at every Primary-verdict publish point. - SqlServer v1: the updater throws NotSupportedException on every arm; the node manager maps it to a node-level BadHistoryOperationUnsupported (was per-item under a Good node for UpdateData/DeleteAtTime). HistoryWrite bit still advertised on v1 (not knowable at materialise time) — documented. - The update plan is evaluated before the Primary gate: an unservable request (Remove, IsDeleteModified, not historized, no updater) is BadHistoryOperationUnsupported off-Primary, not BadInvalidState. - Local-kind SourceTimestamp converted to UTC before the future-skew compare and sent as UTC. - Hosted-service wiring extracted to BuildHistoryUpdatePosture and tested (gate source, skew, clock). - Docs: Historian.md (boot window, ordering, node-level v1, HistoryWrite bit), CLAUDE.md. Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -1347,7 +1347,7 @@ historian tagname, which is the tag's driver `FullName`.
|
||||
- **QuestDB** (`Driver.Historian.QuestDb`, `ServerHistorian:QuestDb:*`, Npgsql over PG wire; docker-dev overlay `docker-compose.questdb.yml`): same kinds/status/aggregates/HistoryUpdate as SqlServer, soft delete by tombstone (INSERT only — never `UPDATE`/`DELETE`; every read filters `NOT deleted`). ⚠️ **Timestamps are µs** (`QuestDbConnection.Round`/`BindTimestamp`; `IHistorianDataSource.TimestampResolutionTicks` = 10 steps HistoryRead paging), **pre-1970 cannot be stored** (writer drops, HistoryUpdate `BadOutOfRange`), every write waits for WAL visibility, TTL is enforced on write commits only, set Npgsql `Maximum Pool Size` < QuestDB's 64. `docs/Historian.md` §"QuestDB backend".
|
||||
- **Real status fix:** the recorder used to stamp every sample Good (`192`); it now carries the driver's status. ⚠️ On the **Gateway** backend this changes nothing stored: the gateway discards quality (C-002), so AVEVA still stores Good.
|
||||
- **Comms loss is historized as an outage marker** (`Empty` + `BadNoCommunication`, one per tag per outage) on SqlServer; dropped (`DroppedUnsupportedKind`) on Gateway. Outbox records are **format v2** — drain or delete `ContinuousHistorization:OutboxPath` before downgrading a node.
|
||||
- **HistoryUpdate** (SqlServer + QuestDb; Gateway ⇒ `BadHistoryOperationUnsupported`): Insert/Replace/Update/DeleteRaw/DeleteAtTime, **Primary-only** since #674 (off-Primary ⇒ `BadInvalidState`, after authorization, same `IDriverPrimaryRoleSource` verdict as the native-ack gate; reads ungated), items with a `SourceTimestamp` past now + `ServerHistorian:HistoryUpdateMaxFutureSkew` (default 1 day, ≤0 disables) ⇒ `BadOutOfRange` in the node manager (every backend), gated in both ACL modes — flag off needs the `HistoryUpdate` role from a **Cluster-scoped** grant (+ key re-check), flag on uses `DataPlaneAuthorizer`'s HistoryUpdate operation; `WriteOperate` never implies it. The `HistoryWrite` AccessLevel bit is set only when an updater is wired at materialise time. `docs/Historian.md` §HistoryUpdate, `docs/security.md`. ⚠️ **SqlServer schema v1 is accepted (#674)** with `AutoCreateSchema=false`: writes/reads/events/retention work; modified reads + HistoryUpdate ⇒ `BadHistoryOperationUnsupported` (nothing sent), retention skips `SampleModified`, one Warning per boot (`SqlServerSchemaBootstrapper.SupportsModifiedTrail`; `RequiredVersion` = 1, `ModifiedTrailVersion` = 2). QuestDB: keep `cairo.ttl.use.wall.clock=true`; a QuestDB trail INSERT retried after a client timeout duplicates (no DEDUP key).
|
||||
- **HistoryUpdate** (SqlServer + QuestDb; Gateway ⇒ `BadHistoryOperationUnsupported`): Insert/Replace/Update/DeleteRaw/DeleteAtTime, **Primary-only** since #674 (off-Primary ⇒ `BadInvalidState`, after authorization AND after planning — an unservable request (Remove / `IsDeleteModified` / not historized) stays `BadHistoryOperationUnsupported` off-Primary; the verdict is `IHistoryUpdateRoleView` = the data-plane verdict but **closed until this node's first redundancy snapshot**, so a pair cannot both accept in the boot window; admin-only nodes register no view ⇒ no gate; reads ungated), items with a `SourceTimestamp` past now + `ServerHistorian:HistoryUpdateMaxFutureSkew` (default 1 day, ≤0 disables) ⇒ `BadOutOfRange` in the node manager (every backend), gated in both ACL modes — flag off needs the `HistoryUpdate` role from a **Cluster-scoped** grant (+ key re-check), flag on uses `DataPlaneAuthorizer`'s HistoryUpdate operation; `WriteOperate` never implies it. The `HistoryWrite` AccessLevel bit is set only when an updater is wired at materialise time. `docs/Historian.md` §HistoryUpdate, `docs/security.md`. ⚠️ **SqlServer schema v1 is accepted (#674)** with `AutoCreateSchema=false`: writes/reads/events/retention work; modified reads + HistoryUpdate ⇒ node-level `BadHistoryOperationUnsupported` on every arm (the updater throws `NotSupportedException`, nothing sent; the `HistoryWrite` bit is still advertised — not knowable at materialise time), retention skips `SampleModified`, one Warning per boot (`SqlServerSchemaBootstrapper.SupportsModifiedTrail`; `RequiredVersion` = 1, `ModifiedTrailVersion` = 2). QuestDB: keep `cairo.ttl.use.wall.clock=true`; a QuestDB trail INSERT retried after a client timeout duplicates (no DEDUP key).
|
||||
- **Modified-value history (#668, Part 11):** every applied HistoryUpdate item leaves a trail row (prior value, principal from the session's API key / `anonymous`, `HistoryModificationType`, time) — SqlServer `hist.SampleModified` (**schema v2**, written by `OUTPUT … INTO` in the same statement ⇒ atomic; ⚠️ no FK/CHECK/trigger on it, and `AutoCreateSchema=false` installs must run `002_sample_modified.sql` before upgrading), QuestDB `{p}sample_modified` (append-only, written **before** the change ⇒ a crash leaves an orphan trail row, never an untracked change). `IsReadModified=true` HistoryRead serves it as `HistoryModifiedData` via `IHistorianDataSource.ReadModifiedAsync` (default ⇒ `NotSupported` ⇒ `BadHistoryOperationUnsupported` on Gateway/Null), paged on `(SourceTime, ModifiedId)`, gated by plain HistoryRead. Live writes and retention never write it; retention/TTL purge it on source time. `IHistorianDataUpdater` methods take a `HistoryUpdateContext`. CLI `historyread --modified`. `docs/Historian.md` §Modified-value history.
|
||||
|
||||
### Read path (`ServerHistorian` section)
|
||||
|
||||
+22
-6
@@ -690,8 +690,11 @@ modification time and the Part 11 `HistoryUpdateType`. A HistoryReadRaw with `Is
|
||||
per tag alongside `hist.Sample`; QuestDB puts the same TTL on `{p}sample_modified`.
|
||||
- **SqlServer schema v1 (#674).** With `AutoCreateSchema=false` a database still at schema **version 1** (script `002_sample_modified.sql` not
|
||||
applied) keeps historizing — live writes, raw/processed/at-time reads, alarm events and retention all work — but has no trail: modified reads
|
||||
and every HistoryUpdate item answer **`BadHistoryOperationUnsupported`** with nothing sent to SQL, retention skips `SampleModified`, and the
|
||||
node logs **one Warning per boot**. Apply `002_sample_modified.sql` and restart the node to enable them (the verdict is taken once per
|
||||
and every HistoryUpdate request answer a **node-level `BadHistoryOperationUnsupported`** (all three arms alike — the updater throws
|
||||
`NotSupportedException`, as the read side does) with nothing sent to SQL, retention skips `SampleModified`, and the
|
||||
node logs **one Warning per boot**. ⚠️ Historized nodes still advertise the `HistoryWrite` AccessLevel bit on a v1 database: the bit is
|
||||
decided at materialise time from "an updater is wired", and whether the schema has the trail is only known after the backend's lazy
|
||||
first-use bootstrap (I/O the node manager does not do while materialising), so a v1 node is admitted to HistoryUpdate and then refused. Apply `002_sample_modified.sql` and restart the node to enable them (the verdict is taken once per
|
||||
process). `AutoCreateSchema=true` (the default) still upgrades v1 to v2 in place. See [Schema](#schema).
|
||||
- **QuestDB retry duplicates (#674).** `{p}sample_modified` is append-only with **no DEDUP key** (a key would collapse genuinely distinct
|
||||
modifications of one sample), and each trail row gets a fresh `modified_id`. So a HistoryUpdate a client **retries** after a timeout — when the
|
||||
@@ -1289,7 +1292,7 @@ mixed (Part 11 reports item outcomes in `OperationResults`); it is Bad only for
|
||||
|
||||
**Per-item rejections** (answered without touching SQL; other items still proceed): a null item ⇒ `BadInvalidArgument`;
|
||||
`SourceTimestamp == DateTime.MinValue` ⇒ `BadInvalidTimestamp`; a **`SourceTimestamp` later than `now + ServerHistorian:HistoryUpdateMaxFutureSkew`**
|
||||
(default 1 day, server clock; #674) ⇒ **`BadOutOfRange`** — on every backend, never sent; a value the historian cannot type (array, unsupported CLR type) ⇒
|
||||
(default 1 day, server clock; #674 — a `Local`-kind timestamp is converted to UTC first, and is sent as UTC) ⇒ **`BadOutOfRange`** — on every backend, never sent; a value the historian cannot type (array, unsupported CLR type) ⇒
|
||||
`BadTypeMismatch`; a **`NaN` / `±Infinity` `Double` ⇒ `BadInvalidArgument`** — it cannot live in a SQL `float`, and unlike the live-value
|
||||
writer (which stores it as `Empty`) an operator's explicit update is rejected rather than silently rewritten; and a **null value with a
|
||||
Good or Uncertain status ⇒ `BadInvalidArgument`** — an `Empty` sample is always Bad, so a null value is accepted only with a Bad status
|
||||
@@ -1312,15 +1315,28 @@ outcome, later ones `BadNoEntryExists`. **Unknown tag:** Insert / Update create
|
||||
`ServerHistorian:Enabled=false`); `PerformUpdateType.Remove`; `DeleteRawModifiedDetails.IsDeleteModified=true` (the
|
||||
[modification trail](#modified-value-history-part-11) is an audit record, aged out only by retention); and the out-of-scope detail types — **event** HistoryUpdate (`UpdateEventDetails` / `DeleteEventDetails`) and `UpdateStructureData` —
|
||||
which are not overridden, so the SDK base rejects them. A backend exception is `BadInternalError` for that node and a deadline is `BadTimeout`.
|
||||
On a `SqlServer` database still at schema v1 every item is `BadHistoryOperationUnsupported` ([#674](#modified-value-history-part-11)).
|
||||
On a `SqlServer` database still at schema v1 the whole node is `BadHistoryOperationUnsupported` on every arm ([#674](#modified-value-history-part-11)).
|
||||
No `AuditHistoryUpdateEvent` is emitted (#674.5, not implemented).
|
||||
|
||||
### Primary-only (#674)
|
||||
|
||||
HistoryUpdate is served only by the node holding the redundancy **Primary** data-plane role — the same verdict
|
||||
(`IDriverPrimaryRoleSource`, published by `DriverHostActor`) that gates native-alarm acknowledges (#597). On any other node every node of
|
||||
the request that **passed authorization** is refused **`BadInvalidState`** (node-level code, with a message pointing at the Primary) and
|
||||
nothing reaches the backend. Authorization runs first, so an unauthorized caller learns `BadUserAccessDenied`, not the node's role.
|
||||
the request that **passed authorization and is servable** is refused **`BadInvalidState`** (node-level code, with a message pointing at the
|
||||
Primary) and nothing reaches the backend. Order: authorization first (an unauthorized caller learns `BadUserAccessDenied`, not the node's
|
||||
role), then the request is **planned** (a request no server could serve — no updater, a node that is not historized, `Remove`,
|
||||
`IsDeleteModified` — is the honest `BadHistoryOperationUnsupported` on every node, so a Secondary does not send the client to a Primary
|
||||
that would refuse it too), then the gate.
|
||||
|
||||
**Closed until the first redundancy snapshot (boot window).** The gate reads its own verdict, `IHistoryUpdateRoleView.ShouldServeHistoryUpdate`
|
||||
(`PrimaryGatePolicy.ShouldServeHistoryUpdate`): the data-plane Primary verdict, but **closed until this node has received its first redundancy
|
||||
snapshot**. The data-plane verdict itself allows during the pre-join boot window (a node not yet joined sees no driver peer, which reads as
|
||||
"single node") — right for live writes and acks, which must not stall on a genuine single node, but it would let both nodes of a pair accept a
|
||||
HistoryUpdate for a few seconds (up to ~25 s under the bootstrap guard). "No verdict yet" is told apart from "single node" by the snapshot
|
||||
itself: the redundancy singleton runs on every driver node — a lone node included — and publishes on every change and every 10 s heartbeat,
|
||||
so a single node is named Primary within seconds of joining and opens then. Until that first snapshot every node refuses `BadInvalidState`;
|
||||
after it the gate is exactly the data-plane verdict (including the unknown-role / member-count rule for a snapshot that omits this node). A
|
||||
node with no redundancy manager (admin-only — it registers no `IHistoryUpdateRoleView`) wires no gate and allows.
|
||||
HistoryRead is **not** gated. Why: rewriting history is a single-writer operation — both backends capture the prior value for the
|
||||
[modification trail](#modified-value-history-part-11) under a per-tag, **per-process** lock (QuestDB) or statement (SqlServer), so two pair
|
||||
nodes updating one tag concurrently could record a wrong prior value. A single-node deployment (no redundancy manager) is unaffected. Clients
|
||||
|
||||
@@ -23,6 +23,12 @@ public enum HistoryUpdateMode
|
||||
/// <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.
|
||||
/// <para>
|
||||
/// <b>Cannot update at all</b> (Gitea #674): an implementation that is wired but cannot perform any update right
|
||||
/// now — the SqlServer backend on a schema without the modification trail — throws
|
||||
/// <see cref="NotSupportedException"/> from every method without touching storage. The server reports it as a
|
||||
/// node-level <c>BadHistoryOperationUnsupported</c>, the same as <see cref="IHistorianDataSource.ReadModifiedAsync"/>.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public interface IHistorianDataUpdater
|
||||
{
|
||||
|
||||
@@ -27,10 +27,4 @@ public static class HistorianStatusCodes
|
||||
/// status, or a tag name longer than <see cref="HistorianTagNames.MaxLength"/>.
|
||||
/// </summary>
|
||||
public const uint BadInvalidArgument = 0x80AB0000u;
|
||||
|
||||
/// <summary>
|
||||
/// A modified-value read or HistoryUpdate item against a SQL Server schema still at version 1 (no Part 11
|
||||
/// modification trail, Gitea #674): refused without touching the database.
|
||||
/// </summary>
|
||||
public const uint BadHistoryOperationUnsupported = 0x80720000u;
|
||||
}
|
||||
|
||||
+16
-16
@@ -28,8 +28,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
|
||||
/// <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>
|
||||
/// <para><b>Schema v1 (Gitea #674).</b> Without the trail (<see cref="SqlServerSchemaBootstrapper.SupportsModifiedTrail"/>
|
||||
/// false) a change could not be audited, so every item (and DeleteRaw's single result) reports
|
||||
/// <c>BadHistoryOperationUnsupported</c> and nothing is sent to the database beyond the one-time schema check.</para>
|
||||
/// false) a change could not be audited, so every call throws <see cref="NotSupportedException"/> — which the node
|
||||
/// manager reports as a node-level <c>BadHistoryOperationUnsupported</c>, on all three arms alike — and nothing is
|
||||
/// sent to the database beyond the one-time schema check.</para>
|
||||
/// </summary>
|
||||
internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
|
||||
{
|
||||
@@ -49,20 +50,19 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
|
||||
_tags = tags;
|
||||
}
|
||||
|
||||
/// <summary>#674: ensures the schema and reports whether the Part 11 trail exists (no SQL after the first success).</summary>
|
||||
/// <summary>
|
||||
/// #674: ensures the schema and refuses the call when it has no Part 11 trail (no SQL after the first success).
|
||||
/// </summary>
|
||||
/// <param name="ct">Cancellation.</param>
|
||||
/// <returns>True when the modification trail is available.</returns>
|
||||
private async Task<bool> TrailAvailableAsync(CancellationToken ct)
|
||||
/// <returns>A task completing when the trail is available.</returns>
|
||||
/// <exception cref="NotSupportedException">The schema is below <see cref="SqlServerSchemaBootstrapper.ModifiedTrailVersion"/>.</exception>
|
||||
private async Task RequireTrailAsync(CancellationToken ct)
|
||||
{
|
||||
await _bootstrap.EnsureAsync(ct).ConfigureAwait(false);
|
||||
return _bootstrap.SupportsModifiedTrail;
|
||||
}
|
||||
|
||||
private static uint[] Unsupported(int count)
|
||||
{
|
||||
var codes = new uint[count];
|
||||
Array.Fill(codes, HistorianStatusCodes.BadHistoryOperationUnsupported);
|
||||
return codes;
|
||||
if (!_bootstrap.SupportsModifiedTrail)
|
||||
throw new NotSupportedException(
|
||||
$"Historian schema [{_conn.Schema}] is below version {SqlServerSchemaBootstrapper.ModifiedTrailVersion}: " +
|
||||
"HistoryUpdate needs the Part 11 modification trail (apply 002_sample_modified.sql and restart).");
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
@@ -72,7 +72,7 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
|
||||
ArgumentNullException.ThrowIfNull(tag);
|
||||
ArgumentNullException.ThrowIfNull(values);
|
||||
ArgumentNullException.ThrowIfNull(context);
|
||||
if (!await TrailAvailableAsync(ct).ConfigureAwait(false)) return Unsupported(values.Count);
|
||||
await RequireTrailAsync(ct).ConfigureAwait(false);
|
||||
var plan = HistoryUpdatePlanner.Plan(tag, mode, values);
|
||||
if (plan.Winners.Count == 0) return HistoryUpdatePlanner.Complete(plan, mode, new Dictionary<long, UpsertOutcome>());
|
||||
|
||||
@@ -92,7 +92,7 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(tag);
|
||||
ArgumentNullException.ThrowIfNull(context);
|
||||
if (!await TrailAvailableAsync(ct).ConfigureAwait(false)) return HistorianStatusCodes.BadHistoryOperationUnsupported;
|
||||
await RequireTrailAsync(ct).ConfigureAwait(false);
|
||||
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);
|
||||
@@ -120,7 +120,7 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
|
||||
ArgumentNullException.ThrowIfNull(tag);
|
||||
ArgumentNullException.ThrowIfNull(timesUtc);
|
||||
ArgumentNullException.ThrowIfNull(context);
|
||||
if (!await TrailAvailableAsync(ct).ConfigureAwait(false)) return Unsupported(timesUtc.Count);
|
||||
await RequireTrailAsync(ct).ConfigureAwait(false);
|
||||
if (!HistorianTagNames.IsValid(tag))
|
||||
{
|
||||
var refused = new uint[timesUtc.Count];
|
||||
|
||||
@@ -13,6 +13,7 @@ using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.OpcUa;
|
||||
@@ -46,6 +47,7 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
private readonly ILogger<OtOpcUaServerHostedService> _logger;
|
||||
private readonly IDriverPrimaryRoleSource? _primaryRoleSource;
|
||||
private readonly HistorianUpdaterHolder _historianUpdater;
|
||||
private readonly IHistoryUpdateRoleView? _historyUpdateRoleView;
|
||||
|
||||
private OpcUaApplicationHost? _appHost;
|
||||
private OtOpcUaSdkServer? _server;
|
||||
@@ -98,6 +100,9 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
/// <param name="historianUpdater">The selected historian backend's optional HistoryUpdate capability.
|
||||
/// Optional: when unregistered it resolves to <see cref="HistorianUpdaterHolder.None"/> and HistoryUpdate is
|
||||
/// reported unsupported on every node.</param>
|
||||
/// <param name="historyUpdateRoleView">Gitea #674 — the HistoryUpdate verdict <c>DriverHostActor</c> publishes (the
|
||||
/// data-plane Primary verdict, closed until this node's first redundancy snapshot). Optional: registered on every
|
||||
/// driver-role node; an admin-only node has none and wires no HistoryUpdate Primary gate (allow).</param>
|
||||
public OtOpcUaServerHostedService(
|
||||
IOptions<OpcUaApplicationHostOptions> options,
|
||||
DeferredAddressSpaceSink deferredSink,
|
||||
@@ -113,7 +118,8 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
TimeProvider time,
|
||||
ILoggerFactory loggerFactory,
|
||||
IDriverPrimaryRoleSource? primaryRoleSource = null,
|
||||
HistorianUpdaterHolder? historianUpdater = null)
|
||||
HistorianUpdaterHolder? historianUpdater = null,
|
||||
IHistoryUpdateRoleView? historyUpdateRoleView = null)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(aclOptions);
|
||||
ArgumentNullException.ThrowIfNull(time);
|
||||
@@ -137,6 +143,7 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
_loggerFactory = loggerFactory;
|
||||
_primaryRoleSource = primaryRoleSource;
|
||||
_historianUpdater = historianUpdater ?? HistorianUpdaterHolder.None;
|
||||
_historyUpdateRoleView = historyUpdateRoleView;
|
||||
_logger = loggerFactory.CreateLogger<OtOpcUaServerHostedService>();
|
||||
}
|
||||
|
||||
@@ -252,15 +259,13 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
if (_primaryRoleSource is { } primaryRoleSource)
|
||||
{
|
||||
_server.SetNativeAlarmAckPrimaryGate(() => primaryRoleSource.ShouldActAsPrimary);
|
||||
// Gitea #674: HistoryUpdate is single-writer — only the Primary rewrites history (BadInvalidState elsewhere).
|
||||
_server.SetHistoryUpdatePrimaryGate(() => primaryRoleSource.ShouldActAsPrimary);
|
||||
}
|
||||
else
|
||||
{
|
||||
// Every driver node registers one (AddOtOpcUaRuntime), so this is a wiring regression, not a mode:
|
||||
// say so rather than let a Secondary silently go back to accepting acks it will revert.
|
||||
_logger.LogWarning(
|
||||
"OtOpcUaServerHostedService: no IDriverPrimaryRoleSource registered — native-alarm acknowledges and HistoryUpdate are accepted on every node regardless of redundancy role");
|
||||
"OtOpcUaServerHostedService: no IDriverPrimaryRoleSource registered — native-alarm acknowledges are accepted on every node regardless of redundancy role");
|
||||
}
|
||||
|
||||
// Wire the reverse-path inbound operator-write gateway: a client write to a writable equipment-tag
|
||||
@@ -319,10 +324,13 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
_server.NodeManager.MaxConcurrentHistoryReadBatches = _serverHistorianOptions.MaxConcurrentHistoryReadBatches;
|
||||
_server.NodeManager.HistoryReadDeadline = _serverHistorianOptions.HistoryReadDeadline;
|
||||
|
||||
// Gitea #674: HistoryUpdate refuses a SourceTimestamp beyond now + HistoryUpdateMaxFutureSkew (BadOutOfRange),
|
||||
// Gitea #674: HistoryUpdate is Primary-only (BadInvalidState elsewhere, and closed until this node's first
|
||||
// redundancy snapshot) and refuses a SourceTimestamp beyond now + HistoryUpdateMaxFutureSkew (BadOutOfRange),
|
||||
// judged against the host clock.
|
||||
_server.NodeManager.HistoryUpdateMaxFutureSkew = _serverHistorianOptions.HistoryUpdateMaxFutureSkew;
|
||||
_server.NodeManager.Clock = _time;
|
||||
var historyUpdate = BuildHistoryUpdatePosture();
|
||||
_server.NodeManager.HistoryUpdatePrimaryGate = historyUpdate.PrimaryGate;
|
||||
_server.NodeManager.HistoryUpdateMaxFutureSkew = historyUpdate.MaxFutureSkew;
|
||||
_server.NodeManager.Clock = historyUpdate.Clock;
|
||||
|
||||
// ServiceLevel publisher needs IServerInternal — only available after Start.
|
||||
if (_server.CurrentInstance is { } serverInternal)
|
||||
@@ -339,6 +347,28 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
_logger.LogInformation("OtOpcUaServerHostedService: SDK started, address-space + ServiceLevel sinks bound");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #674 — what <c>StartAsync</c> wires onto the node manager for HistoryUpdate: the Primary gate (read from
|
||||
/// the <see cref="IHistoryUpdateRoleView"/> on every call; null when none is registered, i.e. allow), the
|
||||
/// future-timestamp bound from <c>ServerHistorian:HistoryUpdateMaxFutureSkew</c>, and the host clock it is judged
|
||||
/// against. Separate from <c>StartAsync</c> so the wiring is testable without booting the SDK.
|
||||
/// </summary>
|
||||
/// <returns>The posture to apply.</returns>
|
||||
internal HistoryUpdatePosture BuildHistoryUpdatePosture()
|
||||
{
|
||||
Func<bool>? gate = _historyUpdateRoleView is { } view ? () => view.ShouldServeHistoryUpdate : null;
|
||||
if (gate is null)
|
||||
_logger.LogInformation(
|
||||
"OtOpcUaServerHostedService: no IHistoryUpdateRoleView registered (no redundancy manager on this node) — HistoryUpdate is not Primary-gated");
|
||||
return new HistoryUpdatePosture(gate, _serverHistorianOptions.HistoryUpdateMaxFutureSkew, _time);
|
||||
}
|
||||
|
||||
/// <summary>Gitea #674 — the HistoryUpdate settings <see cref="BuildHistoryUpdatePosture"/> produces.</summary>
|
||||
/// <param name="PrimaryGate">The node manager's HistoryUpdate Primary gate; null allows.</param>
|
||||
/// <param name="MaxFutureSkew">The future-timestamp bound; non-positive disables it.</param>
|
||||
/// <param name="Clock">The clock the bound is judged against.</param>
|
||||
internal sealed record HistoryUpdatePosture(Func<bool>? PrimaryGate, TimeSpan MaxFutureSkew, TimeProvider Clock);
|
||||
|
||||
/// <summary>
|
||||
/// Arms the clock-driven monitored-item revalidation tick (arch-review 03/S19).
|
||||
/// </summary>
|
||||
|
||||
@@ -5484,6 +5484,20 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
});
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #674 review: a HistoryUpdate timestamp as UTC. Wire timestamps are UTC; an in-process caller can hand a
|
||||
/// <see cref="DateTimeKind.Local"/> value, which is converted (comparing its raw ticks against a UTC bound would be
|
||||
/// off by the zone offset). <see cref="DateTimeKind.Unspecified"/> is treated as UTC, as the backends treat it.
|
||||
/// </summary>
|
||||
/// <param name="value">The item's source timestamp.</param>
|
||||
/// <returns>The same instant as a UTC <see cref="DateTime"/>.</returns>
|
||||
private static DateTime ToUtc(DateTime value) => value.Kind switch
|
||||
{
|
||||
DateTimeKind.Local => value.ToUniversalTime(),
|
||||
DateTimeKind.Unspecified => DateTime.SpecifyKind(value, DateTimeKind.Utc),
|
||||
_ => value,
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// #674: the latest <c>SourceTimestamp</c> a HistoryUpdate item may carry — <see cref="Clock"/>'s now plus
|
||||
/// <see cref="HistoryUpdateMaxFutureSkew"/> — or <c>null</c> when the bound is disabled (non-positive skew).
|
||||
@@ -5533,9 +5547,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
{
|
||||
codes[i] = StatusCodes.BadInvalidTimestamp;
|
||||
}
|
||||
else if (latest is { } bound && item.SourceTimestamp > bound)
|
||||
else if (latest is { } bound && ToUtc(item.SourceTimestamp) > bound)
|
||||
{
|
||||
// #674: a far-future history edit is refused before any backend sees it (wire timestamps are UTC).
|
||||
// #674: a far-future history edit is refused before any backend sees it.
|
||||
codes[i] = StatusCodes.BadOutOfRange;
|
||||
}
|
||||
else if (!HistorianValue.TryFromObject(item.Value, out var value))
|
||||
@@ -5544,7 +5558,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
}
|
||||
else
|
||||
{
|
||||
toSend.Add(new HistorizationValue(item.SourceTimestamp, value, item.StatusCode.Code));
|
||||
toSend.Add(new HistorizationValue(ToUtc(item.SourceTimestamp), value, item.StatusCode.Code));
|
||||
sentIndex.Add(i);
|
||||
}
|
||||
}
|
||||
@@ -5588,13 +5602,35 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
nodesToProcess = AuthorizeHistoryUpdate(context, nodesToProcess, results, errors);
|
||||
if (nodesToProcess.Count == 0) return;
|
||||
|
||||
// #674 — PRIMARY GATE, after authorization (a caller who may not update learns nothing about the node's role).
|
||||
// Off-Primary every authorized node is refused BadInvalidState and nothing reaches the updater.
|
||||
// Plan BEFORE the Primary gate (#674 review): a request this server can never serve (no updater, a node that is
|
||||
// not historized, Remove, IsDeleteModified) is answered with the honest BadHistoryOperationUnsupported on every
|
||||
// node, so a Secondary does not send the client to the Primary for something the Primary would also refuse.
|
||||
// Planning is pure — it builds closures and reads the in-memory tagname map; nothing reaches a backend.
|
||||
var updater = _historianDataUpdater;
|
||||
var planned = new List<(NodeHandle Handle, string Tagname,
|
||||
Func<IHistorianDataUpdater, string, HistoryUpdateContext, CancellationToken, Task<HistoryUpdateResult>> Operation)>(nodesToProcess.Count);
|
||||
foreach (var handle in nodesToProcess)
|
||||
{
|
||||
var operation = updater is null ? null : plan(handle);
|
||||
if (operation is null || !TryGetHistorizedTagname(handle.NodeId, out var tagname) || tagname is null)
|
||||
{
|
||||
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
|
||||
results[handle.Index] = new HistoryUpdateResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported };
|
||||
continue;
|
||||
}
|
||||
|
||||
planned.Add((handle, tagname, operation));
|
||||
}
|
||||
|
||||
if (planned.Count == 0) return;
|
||||
|
||||
// #674 — PRIMARY GATE, after authorization (a caller who may not update learns nothing about the node's role)
|
||||
// and after planning. Off-Primary every servable node is refused BadInvalidState and nothing reaches the updater.
|
||||
if (HistoryUpdatePrimaryGate is { } primary && !primary())
|
||||
{
|
||||
var refusal = new ServiceResult(StatusCodes.BadInvalidState,
|
||||
new LocalizedText("This server is not the redundancy Primary; update history on the Primary (the server with the higher ServiceLevel)."));
|
||||
foreach (var handle in nodesToProcess)
|
||||
foreach (var (handle, _, _) in planned)
|
||||
{
|
||||
errors[handle.Index] = refusal;
|
||||
results[handle.Index] = new HistoryUpdateResult { StatusCode = StatusCodes.BadInvalidState };
|
||||
@@ -5606,25 +5642,17 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
// Who is rewriting history — recorded on every modification-trail row (Part 11 ModificationInfo.UserName, #668).
|
||||
var who = HistoryUpdateCallerOf(context);
|
||||
|
||||
WithHistoryReadBatchLimiter(nodesToProcess, results, errors, () =>
|
||||
WithHistoryReadBatchLimiter(planned.Select(p => p.Handle).ToList(), results, errors, () =>
|
||||
{
|
||||
using var deadline = CreateDeadline();
|
||||
var ct = deadline?.Token ?? CancellationToken.None;
|
||||
var updater = _historianDataUpdater;
|
||||
var work = new List<Func<Task>>(nodesToProcess.Count);
|
||||
foreach (var handle in nodesToProcess)
|
||||
var work = new List<Func<Task>>(planned.Count);
|
||||
foreach (var (handle, tagname, operation) in planned)
|
||||
{
|
||||
var operation = updater is null ? null : plan(handle);
|
||||
if (operation is null || !TryGetHistorizedTagname(handle.NodeId, out var tagname) || tagname is null)
|
||||
{
|
||||
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
|
||||
results[handle.Index] = new HistoryUpdateResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported };
|
||||
continue;
|
||||
}
|
||||
|
||||
var h = handle;
|
||||
var tag = tagname;
|
||||
work.Add(() => ServeUpdateNodeAsync(h, results, errors, token => operation(updater!, tag, who, token), ct));
|
||||
var op = operation;
|
||||
work.Add(() => ServeUpdateNodeAsync(h, results, errors, token => op(updater!, tag, who, token), ct));
|
||||
}
|
||||
|
||||
RunBounded(work, HistoryReadBatchParallelism);
|
||||
@@ -5677,6 +5705,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
errors[handle.Index] = StatusCodes.BadTimeout;
|
||||
results[handle.Index] = new HistoryUpdateResult { StatusCode = StatusCodes.BadTimeout };
|
||||
}
|
||||
catch (NotSupportedException)
|
||||
{
|
||||
// The backend cannot perform this update at all (e.g. a SqlServer historian still at schema v1, with no
|
||||
// modification trail to record it in, #674) — the spec status for the whole node, not a server fault.
|
||||
// Mirrors the HistoryRead side's handling of the same exception.
|
||||
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
|
||||
results[handle.Index] = new HistoryUpdateResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported };
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "OtOpcUaNodeManager: HistoryUpdate failed for node {NodeId}", handle.NodeId);
|
||||
|
||||
@@ -452,6 +452,13 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// </summary>
|
||||
private readonly IHistorizationRoleView? _historizationRoleView;
|
||||
|
||||
/// <summary>Gitea #674 review: where the HistoryUpdate verdict is published. Null on nodes that wire no redundancy manager.</summary>
|
||||
private readonly IHistoryUpdateRoleView? _historyUpdateRoleView;
|
||||
|
||||
/// <summary>Gitea #674 review: whether any redundancy snapshot has reached this host — the HistoryUpdate gate stays
|
||||
/// closed until one has (see <see cref="PrimaryGatePolicy.ShouldServeHistoryUpdate"/>).</summary>
|
||||
private bool _redundancySnapshotReceived;
|
||||
|
||||
/// <summary>
|
||||
/// Tests whether a node named in a redundancy snapshot serves this node's address space — i.e.
|
||||
/// is a per-cluster mesh partner recording the very same historized refs. Null (the default)
|
||||
@@ -689,7 +696,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
TimeSpan? commsRecoveryAuditDelay = null,
|
||||
Func<CommonsNodeId, bool>? sharesAlarmQueueWithLocal = null,
|
||||
TimeSpan? applyOutcomeTimeout = null,
|
||||
bool captureWhilePrimaryUnreachable = true) =>
|
||||
bool captureWhilePrimaryUnreachable = true,
|
||||
IHistoryUpdateRoleView? historyUpdateRoleView = null) =>
|
||||
// WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an
|
||||
// expression tree. Six IActorRef? parameters and several interface-typed ones mean a
|
||||
// mis-ordered argument is usually type-compatible and therefore compiles clean, then binds
|
||||
@@ -702,7 +710,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher,
|
||||
alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView,
|
||||
sharesHistorizationScopeWithLocal, applyWedgeTimeout, reprovisionBackoff, commsRecoveryAuditDelay,
|
||||
sharesAlarmQueueWithLocal, applyOutcomeTimeout, captureWhilePrimaryUnreachable));
|
||||
sharesAlarmQueueWithLocal, applyOutcomeTimeout, captureWhilePrimaryUnreachable, historyUpdateRoleView));
|
||||
|
||||
/// <summary>Initializes a new DriverHostActor with the specified dependencies.</summary>
|
||||
/// <param name="dbFactory">Database context factory for configuration database access.</param>
|
||||
@@ -775,6 +783,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// reported UNREACHABLE by the cluster failure detector, stop deferring to it and capture — rather than waiting
|
||||
/// for auto-down + a fresh redundancy snapshot. Affects ONLY the historization capture verdict; the alarm drain
|
||||
/// and the Primary data-plane verdict are unchanged. <c>false</c> restores the pre-#672 behaviour.</param>
|
||||
/// <param name="historyUpdateRoleView">Gitea #674 review: where the OPC UA HistoryUpdate verdict
|
||||
/// (<see cref="PrimaryGatePolicy.ShouldServeHistoryUpdate"/> — the data-plane verdict, but closed until the first
|
||||
/// redundancy snapshot) is published for the node manager's HistoryUpdate Primary gate. Null on nodes that wire no
|
||||
/// redundancy manager.</param>
|
||||
public DriverHostActor(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory,
|
||||
CommonsNodeId localNode,
|
||||
@@ -807,9 +819,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
TimeSpan? commsRecoveryAuditDelay = null,
|
||||
Func<CommonsNodeId, bool>? sharesAlarmQueueWithLocal = null,
|
||||
TimeSpan? applyOutcomeTimeout = null,
|
||||
bool captureWhilePrimaryUnreachable = true)
|
||||
bool captureWhilePrimaryUnreachable = true,
|
||||
IHistoryUpdateRoleView? historyUpdateRoleView = null)
|
||||
{
|
||||
_captureWhilePrimaryUnreachable = captureWhilePrimaryUnreachable;
|
||||
_historyUpdateRoleView = historyUpdateRoleView;
|
||||
_applyOutcomeTimeout = applyOutcomeTimeout ?? DefaultApplyOutcomeTimeout;
|
||||
_sharesAlarmQueueWithLocal = sharesAlarmQueueWithLocal;
|
||||
_applyWedgeTimeout = applyWedgeTimeout ?? DefaultApplyWedgeTimeout;
|
||||
@@ -909,6 +923,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// </summary>
|
||||
private void StartPrimaryVerdictPublishing()
|
||||
{
|
||||
PublishHistoryUpdateVerdict();
|
||||
if (_primaryRoleSource is null) return;
|
||||
PublishPrimaryVerdict();
|
||||
try
|
||||
@@ -948,6 +963,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// The snapshot path publishes on every snapshot, as the other verdicts do.</summary>
|
||||
private void PublishPrimaryVerdict()
|
||||
{
|
||||
PublishHistoryUpdateVerdict();
|
||||
if (_primaryRoleSource is null) return;
|
||||
var verdict = ShouldServiceAsPrimary();
|
||||
_lastPublishedPrimaryVerdict = verdict;
|
||||
@@ -958,10 +974,18 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// role information, so republishing an unchanged verdict on each would be noise.</summary>
|
||||
private void PublishPrimaryVerdictIfChanged()
|
||||
{
|
||||
PublishHistoryUpdateVerdict();
|
||||
if (_primaryRoleSource is null || ShouldServiceAsPrimary() == _lastPublishedPrimaryVerdict) return;
|
||||
PublishPrimaryVerdict();
|
||||
}
|
||||
|
||||
/// <summary>Gitea #674 review: publishes <see cref="PrimaryGatePolicy.ShouldServeHistoryUpdate"/> (no-op when no view
|
||||
/// is wired). Rides every Primary-verdict publish point — host start, every snapshot, membership changes — so it is
|
||||
/// closed from host start and opens only once a snapshot has arrived and the data-plane verdict allows.</summary>
|
||||
private void PublishHistoryUpdateVerdict() =>
|
||||
_historyUpdateRoleView?.PublishHistoryUpdateRole(
|
||||
PrimaryGatePolicy.ShouldServeHistoryUpdate(_redundancySnapshotReceived, _localRole, DriverMemberCount()));
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #672: follow cluster reachability so the historization capture gate stops deferring to a recording
|
||||
/// Primary the failure detector already considers dead. Its OWN subscription, deliberately not folded into
|
||||
@@ -2270,6 +2294,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// <c>ScriptedAlarmHostActor.OnRedundancyStateChanged</c> / <c>OpcUaPublishActor</c>.</summary>
|
||||
private void OnRedundancyStateChanged(RedundancyStateChanged msg)
|
||||
{
|
||||
_redundancySnapshotReceived = true;
|
||||
var local = msg.Nodes.FirstOrDefault(n => n.NodeId == _localNode);
|
||||
if (local is not null)
|
||||
{
|
||||
|
||||
@@ -26,6 +26,36 @@ public static class PrimaryGatePolicy
|
||||
_ => driverMemberCount <= 1, // unknown: allow only when no driver peer exists
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #674 review: decide whether this node may serve an OPC UA <b>HistoryUpdate</b>. The
|
||||
/// <see cref="ShouldServiceAsPrimary"/> verdict, but CLOSED until this node has received its first
|
||||
/// redundancy snapshot.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <see cref="ShouldServiceAsPrimary"/> deliberately allows an unknown role while no driver peer is
|
||||
/// visible, which covers the pre-join boot window: a node spawned before it has joined its mesh sees a
|
||||
/// member count of 0. For a live data-plane operation (a device write, an ack) that is the right trade —
|
||||
/// denying would stall a genuine single node during its own join. HistoryUpdate is not live: a client can
|
||||
/// retry it, and both nodes of a pair accepting one concurrently during that window is exactly the
|
||||
/// wrong-prior-value race the Primary gate exists for. So it waits for a real verdict.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// "No snapshot yet" is told apart from "single node" by the snapshot itself: the redundancy singleton runs
|
||||
/// on every driver node (one per mesh, a single node included) and publishes on every change and every
|
||||
/// 10 s heartbeat, so a single node is named Primary within seconds of joining. Once a snapshot has
|
||||
/// arrived this is exactly <see cref="ShouldServiceAsPrimary"/> (including its unknown-role member-count
|
||||
/// rule for a snapshot that omits this node). A node with no redundancy manager at all (admin-only, or a
|
||||
/// harness) wires no gate and so allows.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="redundancySnapshotReceived">Whether any redundancy snapshot has reached this node yet.</param>
|
||||
/// <param name="localRole">This node's last-known redundancy role, or <c>null</c> when unknown.</param>
|
||||
/// <param name="driverMemberCount">Count of Up cluster members carrying the <c>driver</c> role.</param>
|
||||
/// <returns><c>true</c> to serve HistoryUpdate; <c>false</c> to refuse it (<c>BadInvalidState</c>).</returns>
|
||||
public static bool ShouldServeHistoryUpdate(bool redundancySnapshotReceived, RedundancyRole? localRole, int driverMemberCount) =>
|
||||
redundancySnapshotReceived && ShouldServiceAsPrimary(localRole, driverMemberCount);
|
||||
|
||||
/// <summary>
|
||||
/// Decide whether this node should drain the replicated alarm store-and-forward queue.
|
||||
/// </summary>
|
||||
|
||||
@@ -75,9 +75,29 @@ public interface IHistorizationRoleView
|
||||
void PublishHistorizationRole(bool shouldRecordHistorizedValues);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #674 review: a singleton snapshot of the OPC UA <b>HistoryUpdate</b> verdict
|
||||
/// (<see cref="PrimaryGatePolicy.ShouldServeHistoryUpdate"/>), read by the node manager's HistoryUpdate Primary gate.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// A separate verdict from <see cref="IDriverPrimaryRoleSource.ShouldActAsPrimary"/> because it is closed until the
|
||||
/// first redundancy snapshot, where that one mirrors the live data-plane gate and is open through the pre-join boot
|
||||
/// window. Registered unconditionally on every driver-role node (same object as the other views); a node that does
|
||||
/// not register it wires no gate.
|
||||
/// </remarks>
|
||||
public interface IHistoryUpdateRoleView
|
||||
{
|
||||
/// <summary>Whether this node may serve HistoryUpdate right now.</summary>
|
||||
bool ShouldServeHistoryUpdate { get; }
|
||||
|
||||
/// <summary>Records a fresh decision. Called by <c>DriverHostActor</c> whenever it publishes its Primary verdict.</summary>
|
||||
/// <param name="shouldServeHistoryUpdate">The decision <see cref="PrimaryGatePolicy.ShouldServeHistoryUpdate"/> produced.</param>
|
||||
void PublishHistoryUpdateRole(bool shouldServeHistoryUpdate);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Thread-safe <see cref="IRedundancyRoleView"/> + <see cref="IDriverPrimaryRoleSource"/> +
|
||||
/// <see cref="IHistorizationRoleView"/>, each backed by its own volatile field.
|
||||
/// <see cref="IHistorizationRoleView"/> + <see cref="IHistoryUpdateRoleView"/>, each backed by its own volatile field.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
@@ -90,8 +110,15 @@ public interface IHistorizationRoleView
|
||||
/// them is the bug that live gate already found once.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class RedundancyRoleView : IRedundancyRoleView, IDriverPrimaryRoleSource, IHistorizationRoleView
|
||||
public sealed class RedundancyRoleView : IRedundancyRoleView, IDriverPrimaryRoleSource, IHistorizationRoleView,
|
||||
IHistoryUpdateRoleView
|
||||
{
|
||||
// Gitea #674 review: seeded CLOSED, unlike the data-plane verdict. HistoryUpdate waits for a real redundancy verdict
|
||||
// (PrimaryGatePolicy.ShouldServeHistoryUpdate); a node that never publishes here is not a driver node with a
|
||||
// redundancy manager, and such a node does not register this view as its gate source.
|
||||
private volatile bool _shouldServeHistoryUpdate =
|
||||
PrimaryGatePolicy.ShouldServeHistoryUpdate(redundancySnapshotReceived: false, localRole: null, driverMemberCount: 1);
|
||||
|
||||
// Seeded with the decision for "role unknown" rather than a bare false. An unpublished view and a
|
||||
// node whose role nothing ever reports are genuinely the same situation, and they must behave the
|
||||
// same: a deployment that runs no redundancy at all never publishes here, and defaulting closed
|
||||
@@ -121,6 +148,12 @@ public sealed class RedundancyRoleView : IRedundancyRoleView, IDriverPrimaryRole
|
||||
/// <inheritdoc/>
|
||||
public bool ShouldActAsPrimary => _shouldActAsPrimary;
|
||||
|
||||
/// <inheritdoc/>
|
||||
public bool ShouldServeHistoryUpdate => _shouldServeHistoryUpdate;
|
||||
|
||||
/// <inheritdoc/>
|
||||
public void PublishHistoryUpdateRole(bool shouldServeHistoryUpdate) => _shouldServeHistoryUpdate = shouldServeHistoryUpdate;
|
||||
|
||||
/// <inheritdoc/>
|
||||
public void Publish(bool shouldDrainAlarmHistory) => _shouldDrainAlarmHistory = shouldDrainAlarmHistory;
|
||||
|
||||
|
||||
@@ -98,6 +98,11 @@ public static class ServiceCollectionExtensions
|
||||
// with no gate at all, and both nodes would write every sample twice with nothing to say so.
|
||||
services.TryAddSingleton<IHistorizationRoleView>(sp => sp.GetRequiredService<RedundancyRoleView>());
|
||||
|
||||
// Gitea #674 review: the OPC UA HistoryUpdate verdict — the data-plane verdict, but closed until this node's first
|
||||
// redundancy snapshot. Same instance; registered here (driver-role nodes only) so an admin-only node, which has no
|
||||
// redundancy manager, wires no HistoryUpdate gate at all.
|
||||
services.TryAddSingleton<IHistoryUpdateRoleView>(sp => sp.GetRequiredService<RedundancyRoleView>());
|
||||
|
||||
// The node→central ack transport under MeshTransport:Mode = Grpc (mesh gRPC command plane,
|
||||
// Task 1B.2). Registered unconditionally but resolved only by the Grpc arm of
|
||||
// WithOtOpcUaRuntimeActors — the mode is not knowable here (this runs before configuration is
|
||||
@@ -631,7 +636,9 @@ public static class ServiceCollectionExtensions
|
||||
// Review R6 MINOR-4: the passive half of a LocalDb pair has no PeerAddress, so it is told
|
||||
// its queue-sharing peer as its mesh partner instead — otherwise it never stands down.
|
||||
sharesAlarmQueueWithLocal: BuildAlarmQueuePeerPredicate(
|
||||
resolver.GetService<IConfiguration>(), roleInfo, replicationPeerHost)),
|
||||
resolver.GetService<IConfiguration>(), roleInfo, replicationPeerHost),
|
||||
// Gitea #674 review: the HistoryUpdate verdict (closed until the first redundancy snapshot).
|
||||
historyUpdateRoleView: resolver.GetService<IHistoryUpdateRoleView>()),
|
||||
DriverHostActorName);
|
||||
registry.Register<DriverHostActorKey>(driverHost);
|
||||
|
||||
|
||||
-5
@@ -14,11 +14,6 @@ public sealed class StatusCodeOracleTests
|
||||
((uint)Opc.Ua.AggregateBits.Calculated).ShouldBe(HistorianInfoBits.HistorianCalculated);
|
||||
}
|
||||
|
||||
/// <summary>#674: the v1-schema refusal code is the SDK's BadHistoryOperationUnsupported.</summary>
|
||||
[Fact]
|
||||
public void History_operation_unsupported_matches_the_SDK() =>
|
||||
HistorianStatusCodes.BadHistoryOperationUnsupported.ShouldBe(Opc.Ua.StatusCodes.BadHistoryOperationUnsupported);
|
||||
|
||||
[Fact]
|
||||
public void Interpolated_bits_match_the_SDK()
|
||||
{
|
||||
|
||||
+11
-11
@@ -59,20 +59,20 @@ public sealed class SchemaV1PostureTests(SqlHistorianFixture fx)
|
||||
source.ReadModifiedAsync("Plant.Speed", Recent.AddDays(-1), Recent.AddDays(1), 10, null, Ct));
|
||||
source.GetHealthSnapshot().LastError.ShouldBeNull();
|
||||
|
||||
// HistoryUpdate: every item refused, and NOTHING reaches SQL — a sent MERGE/DELETE would either change the
|
||||
// samples or fail on the missing SampleModified target; an Insert on an unknown tag would create its Tag row.
|
||||
// HistoryUpdate: every arm refused as a whole (NotSupportedException ⇒ the node manager's node-level
|
||||
// BadHistoryOperationUnsupported), and NOTHING reaches SQL — a sent MERGE/DELETE would either change the samples or
|
||||
// fail on the missing SampleModified target (SqlException, not NotSupported); an Insert on an unknown tag would
|
||||
// create its Tag row.
|
||||
var updater = provider.CreateUpdater(sp)!;
|
||||
var tagsBefore = await CountAsync($"SELECT COUNT(*) FROM [{s}].Tag");
|
||||
(await updater.UpdateDataAsync("Plant.Speed", HistoryUpdateMode.Update,
|
||||
await Should.ThrowAsync<NotSupportedException>(() => updater.UpdateDataAsync("Plant.Speed", HistoryUpdateMode.Update,
|
||||
[new HistorizationValue(Recent, HistorianValue.FromDouble(99), 0), new HistorizationValue(Recent.AddSeconds(1), HistorianValue.FromDouble(98), 0)],
|
||||
Who, Ct)).ShouldBe([HistorianStatusCodes.BadHistoryOperationUnsupported, HistorianStatusCodes.BadHistoryOperationUnsupported]);
|
||||
(await updater.UpdateDataAsync("Plant.Brand.New", HistoryUpdateMode.Insert,
|
||||
[new HistorizationValue(Recent, HistorianValue.FromDouble(1), 0)], Who, Ct))
|
||||
.ShouldBe([HistorianStatusCodes.BadHistoryOperationUnsupported]);
|
||||
(await updater.DeleteRawAsync("Plant.Speed", Recent.AddMinutes(-1), Recent.AddMinutes(1), Who, Ct))
|
||||
.ShouldBe(HistorianStatusCodes.BadHistoryOperationUnsupported);
|
||||
(await updater.DeleteAtTimeAsync("Plant.Speed", [Recent], Who, Ct))
|
||||
.ShouldBe([HistorianStatusCodes.BadHistoryOperationUnsupported]);
|
||||
Who, Ct));
|
||||
await Should.ThrowAsync<NotSupportedException>(() => updater.UpdateDataAsync("Plant.Brand.New", HistoryUpdateMode.Insert,
|
||||
[new HistorizationValue(Recent, HistorianValue.FromDouble(1), 0)], Who, Ct));
|
||||
await Should.ThrowAsync<NotSupportedException>(() =>
|
||||
updater.DeleteRawAsync("Plant.Speed", Recent.AddMinutes(-1), Recent.AddMinutes(1), Who, Ct));
|
||||
await Should.ThrowAsync<NotSupportedException>(() => updater.DeleteAtTimeAsync("Plant.Speed", [Recent], Who, Ct));
|
||||
(await CountAsync($"SELECT COUNT(*) FROM [{s}].Tag")).ShouldBe(tagsBefore);
|
||||
(await source.ReadRawAsync("Plant.Speed", Recent.AddMinutes(-1), Recent.AddMinutes(1), 10, Ct))
|
||||
.Samples.Select(v => v.Value).ShouldBe([2d]);
|
||||
|
||||
+116
@@ -0,0 +1,116 @@
|
||||
using Akka.Hosting;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Microsoft.Extensions.Options;
|
||||
using Shouldly;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
|
||||
using ZB.MOM.WW.OtOpcUa.Host.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.Tests.OpcUa;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #674 review — pins what the hosted service wires onto the node manager for HistoryUpdate: the Primary gate
|
||||
/// from <see cref="IHistoryUpdateRoleView"/> (read live on every call), and the future-skew bound + clock from
|
||||
/// <c>ServerHistorian</c> options. Asserted through <see cref="OtOpcUaServerHostedService.BuildHistoryUpdatePosture"/>
|
||||
/// (what <c>StartAsync</c> applies) so the SDK need not boot.
|
||||
/// </summary>
|
||||
public sealed class OtOpcUaServerHostedServiceHistoryUpdateTests
|
||||
{
|
||||
[Fact]
|
||||
public void A_registered_view_wires_a_gate_that_reads_it_live()
|
||||
{
|
||||
var view = new FlippableView { Serve = false };
|
||||
var posture = Service(view).BuildHistoryUpdatePosture();
|
||||
|
||||
posture.PrimaryGate.ShouldNotBeNull();
|
||||
posture.PrimaryGate().ShouldBeFalse();
|
||||
view.Serve = true;
|
||||
posture.PrimaryGate().ShouldBeTrue("the gate must read the view on every call, not capture its value");
|
||||
}
|
||||
|
||||
/// <summary>The gate comes from the HistoryUpdate view, NOT the data-plane source — a source saying "Primary" must not
|
||||
/// open it while the HistoryUpdate verdict is closed (the boot-window distinction).</summary>
|
||||
[Fact]
|
||||
public void The_gate_ignores_the_data_plane_primary_source()
|
||||
{
|
||||
var posture = Service(new FlippableView { Serve = false }, new AlwaysPrimary()).BuildHistoryUpdatePosture();
|
||||
|
||||
posture.PrimaryGate!().ShouldBeFalse();
|
||||
}
|
||||
|
||||
/// <summary>No view (an admin-only node: no redundancy manager) ⇒ no gate ⇒ HistoryUpdate allowed.</summary>
|
||||
[Fact]
|
||||
public void No_view_wires_no_gate()
|
||||
{
|
||||
Service(view: null).BuildHistoryUpdatePosture().PrimaryGate.ShouldBeNull();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Skew_and_clock_come_from_options_and_the_host_clock()
|
||||
{
|
||||
var clock = new FixedClock();
|
||||
var config = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary<string, string?>
|
||||
{
|
||||
["ServerHistorian:HistoryUpdateMaxFutureSkew"] = "00:07:00",
|
||||
}).Build();
|
||||
|
||||
var posture = Service(view: null, config: config, clock: clock).BuildHistoryUpdatePosture();
|
||||
|
||||
posture.MaxFutureSkew.ShouldBe(TimeSpan.FromMinutes(7));
|
||||
posture.Clock.ShouldBeSameAs(clock);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void An_absent_section_keeps_the_one_day_default()
|
||||
{
|
||||
Service(view: null).BuildHistoryUpdatePosture().MaxFutureSkew.ShouldBe(TimeSpan.FromDays(1));
|
||||
}
|
||||
|
||||
private static OtOpcUaServerHostedService Service(
|
||||
IHistoryUpdateRoleView? view, IDriverPrimaryRoleSource? primary = null, IConfiguration? config = null,
|
||||
TimeProvider? clock = null) => new(
|
||||
Options.Create(new OpcUaApplicationHostOptions()),
|
||||
new DeferredAddressSpaceSink(),
|
||||
new DeferredServiceLevelPublisher(),
|
||||
NullOpcUaUserAuthenticator.Instance,
|
||||
NullHistorianDataSource.Instance,
|
||||
EmptyDataPlaneAuthoritySource.Instance,
|
||||
config ?? new ConfigurationBuilder().Build(),
|
||||
() => throw new NotSupportedException("no ActorSystem is needed for the HistoryUpdate posture"),
|
||||
new ActorRegistry(),
|
||||
dataPlaneAuthorizer: null,
|
||||
Options.Create(new AclOptions()),
|
||||
clock ?? TimeProvider.System,
|
||||
NullLoggerFactory.Instance,
|
||||
primaryRoleSource: primary,
|
||||
historyUpdateRoleView: view);
|
||||
|
||||
private sealed class FlippableView : IHistoryUpdateRoleView
|
||||
{
|
||||
public volatile bool Serve;
|
||||
|
||||
public bool ShouldServeHistoryUpdate => Serve;
|
||||
|
||||
public void PublishHistoryUpdateRole(bool shouldServeHistoryUpdate) => Serve = shouldServeHistoryUpdate;
|
||||
}
|
||||
|
||||
private sealed class AlwaysPrimary : IDriverPrimaryRoleSource
|
||||
{
|
||||
public bool ShouldActAsPrimary => true;
|
||||
|
||||
public void PublishPrimaryRole(bool shouldActAsPrimary)
|
||||
{
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class FixedClock : TimeProvider
|
||||
{
|
||||
public override DateTimeOffset GetUtcNow() => new(2026, 10, 11, 0, 0, 0, TimeSpan.Zero);
|
||||
}
|
||||
}
|
||||
@@ -669,7 +669,7 @@ public sealed class HistoryUpdateTests : IDisposable
|
||||
|
||||
var (results, errors) = Insert(nm, identity, nodeId, Value(T1, 1d));
|
||||
var (rawResults, rawErrors) = Call(nm, identity, typeof(DeleteRawModifiedDetails),
|
||||
new DeleteRawModifiedDetails { NodeId = nodeId, StartTime = T1, EndTime = T2 });
|
||||
new DeleteRawModifiedDetails { NodeId = nodeId, StartTime = T1, EndTime = T2, IsDeleteModified = false });
|
||||
var (atResults, atErrors) = Call(nm, identity, typeof(DeleteAtTimeDetails),
|
||||
new DeleteAtTimeDetails { NodeId = nodeId, ReqTimes = new DateTimeCollection { T1 } });
|
||||
|
||||
@@ -742,6 +742,123 @@ public sealed class HistoryUpdateTests : IDisposable
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
// ───────────────────────────── #674 review follow-ups ─────────────────────────────
|
||||
|
||||
/// <summary>
|
||||
/// #674 review: a backend that cannot perform the update at all (the SqlServer updater on a v1 schema throws
|
||||
/// <see cref="NotSupportedException"/>) is a NODE-level <c>BadHistoryOperationUnsupported</c> on all three arms —
|
||||
/// not a per-item code under a Good node, and not <c>BadInternalError</c>.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_not_supported_backend_is_node_level_unsupported_on_every_arm()
|
||||
{
|
||||
var (host, server) = await BootAsync();
|
||||
try
|
||||
{
|
||||
var nm = server.NodeManager!;
|
||||
nm.HistorianDataUpdater = new FakeUpdater { Throw = new NotSupportedException("schema v1: no trail") };
|
||||
WireAcl(nm, enabled: false, NodePermissions.HistoryUpdate);
|
||||
var nodeId = Historized(nm, AddressSpaceRealm.Raw, RawPath);
|
||||
var identity = KeyIdentity(OpcUaDataPlaneRoles.HistoryUpdate);
|
||||
|
||||
var update = Insert(nm, identity, nodeId, Value(T1, 1d));
|
||||
var deleteRaw = Call(nm, identity, typeof(DeleteRawModifiedDetails),
|
||||
new DeleteRawModifiedDetails { NodeId = nodeId, StartTime = T1, EndTime = T2, IsDeleteModified = false });
|
||||
var deleteAt = Call(nm, identity, typeof(DeleteAtTimeDetails),
|
||||
new DeleteAtTimeDetails { NodeId = nodeId, ReqTimes = new DateTimeCollection { T1 } });
|
||||
|
||||
foreach (var (results, errors) in new[] { update, deleteRaw, deleteAt })
|
||||
{
|
||||
errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
|
||||
results[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
|
||||
(results[0].OperationResults?.Count ?? 0).ShouldBe(0);
|
||||
}
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #674 review: a request this server can never serve is answered with the honest
|
||||
/// <c>BadHistoryOperationUnsupported</c> even off-Primary — the plan is evaluated BEFORE the Primary gate, which is
|
||||
/// not even consulted when nothing in the request is servable. A servable node in the same situation still gets
|
||||
/// <c>BadInvalidState</c> (the control: the gate is wired and closed).
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Off_primary_an_unservable_request_is_unsupported_not_invalid_state()
|
||||
{
|
||||
var (host, server) = await BootAsync();
|
||||
try
|
||||
{
|
||||
var nm = server.NodeManager!;
|
||||
var updater = new FakeUpdater();
|
||||
nm.HistorianDataUpdater = updater;
|
||||
var consulted = 0;
|
||||
nm.HistoryUpdatePrimaryGate = () => { Interlocked.Increment(ref consulted); return false; };
|
||||
WireAcl(nm, enabled: false, NodePermissions.HistoryUpdate);
|
||||
var nodeId = Historized(nm, AddressSpaceRealm.Raw, RawPath);
|
||||
var identity = KeyIdentity(OpcUaDataPlaneRoles.HistoryUpdate);
|
||||
|
||||
var (removeResults, removeErrors) = Update(nm, identity, nodeId, PerformUpdateType.Remove, Value(T1, 1d));
|
||||
var (modResults, modErrors) = Call(nm, identity, typeof(DeleteRawModifiedDetails),
|
||||
new DeleteRawModifiedDetails { NodeId = nodeId, StartTime = T1, EndTime = T2, IsDeleteModified = true });
|
||||
|
||||
removeErrors[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
|
||||
removeResults[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
|
||||
modErrors[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
|
||||
modResults[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
|
||||
consulted.ShouldBe(0);
|
||||
|
||||
var (_, servableErrors) = Insert(nm, identity, nodeId, Value(T1, 1d));
|
||||
servableErrors[0].StatusCode.Code.ShouldBe(StatusCodes.BadInvalidState);
|
||||
consulted.ShouldBe(1);
|
||||
updater.UpdateCalls.ShouldBeEmpty();
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #674 review: a <see cref="DateTimeKind.Local"/> SourceTimestamp is converted to UTC before the future-skew
|
||||
/// compare (and before it is sent). Two items one tick either side of the bound, expressed in local time: the one
|
||||
/// past the bound is refused and the one inside it is sent as UTC. Comparing raw ticks would get exactly one of
|
||||
/// them wrong in any non-UTC zone (which one depends on the sign of the offset).
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_local_kind_source_time_is_judged_and_sent_as_utc()
|
||||
{
|
||||
var (host, server) = await BootAsync();
|
||||
try
|
||||
{
|
||||
var nm = server.NodeManager!;
|
||||
var updater = new FakeUpdater();
|
||||
nm.HistorianDataUpdater = updater;
|
||||
nm.Clock = new FakeClock(Start);
|
||||
nm.HistoryUpdateMaxFutureSkew = TimeSpan.FromHours(1);
|
||||
WireAcl(nm, enabled: false, NodePermissions.HistoryUpdate);
|
||||
var nodeId = Historized(nm, AddressSpaceRealm.Raw, RawPath);
|
||||
var bound = Start.AddHours(1);
|
||||
var justPast = bound.AddTicks(1).ToLocalTime();
|
||||
var justInside = bound.AddTicks(-1).ToLocalTime();
|
||||
justPast.Kind.ShouldBe(DateTimeKind.Local);
|
||||
|
||||
var (results, _) = Update(nm, KeyIdentity(OpcUaDataPlaneRoles.HistoryUpdate), nodeId, PerformUpdateType.Update,
|
||||
LocalValue(justPast, 1d), LocalValue(justInside, 2d));
|
||||
|
||||
results[0].OperationResults.Select(c => c.Code)
|
||||
.ShouldBe([StatusCodes.BadOutOfRange, StatusCodes.GoodEntryInserted]);
|
||||
var sent = updater.UpdateCalls.ShouldHaveSingleItem().Values.ShouldHaveSingleItem();
|
||||
sent.TimestampUtc.ShouldBe(bound.AddTicks(-1));
|
||||
sent.TimestampUtc!.Value.Kind.ShouldBe(DateTimeKind.Utc);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
private static DataValue LocalValue(DateTime local, double value)
|
||||
{
|
||||
var dv = new DataValue(new Variant(value)) { StatusCode = StatusCodes.Good };
|
||||
dv.SourceTimestamp = local;
|
||||
return dv;
|
||||
}
|
||||
|
||||
// ───────────────────────────── helpers ─────────────────────────────
|
||||
|
||||
private static DataValue Value(DateTime timestamp, object value) =>
|
||||
@@ -849,6 +966,7 @@ public sealed class HistoryUpdateTests : IDisposable
|
||||
CancellationToken ct)
|
||||
{
|
||||
await Task.Delay(1, ct).ConfigureAwait(false);
|
||||
if (Throw is not null) throw Throw;
|
||||
lock (DeleteRawCalls) { DeleteRawCalls.Add((tag, startUtc, endUtc)); lock (UpdateCalls) Callers.Add(context); }
|
||||
return DeleteRawCode;
|
||||
}
|
||||
@@ -857,6 +975,7 @@ public sealed class HistoryUpdateTests : IDisposable
|
||||
string tag, IReadOnlyList<DateTime> timesUtc, HistoryUpdateContext context, CancellationToken ct)
|
||||
{
|
||||
await Task.Delay(1, ct).ConfigureAwait(false);
|
||||
if (Throw is not null) throw Throw;
|
||||
lock (DeleteAtTimeCalls) { DeleteAtTimeCalls.Add(timesUtc.ToList()); lock (UpdateCalls) Callers.Add(context); }
|
||||
return DeleteAtTimeCodes ?? Enumerable.Repeat(StatusCodes.Good, timesUtc.Count).ToArray();
|
||||
}
|
||||
|
||||
+133
@@ -0,0 +1,133 @@
|
||||
using Akka.Actor;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Types;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #674 review — the OPC UA HistoryUpdate verdict (<see cref="IHistoryUpdateRoleView"/>) is the Primary
|
||||
/// data-plane verdict but CLOSED until this node's first redundancy snapshot, so both nodes of a pair cannot accept a
|
||||
/// HistoryUpdate during the pre-join boot window (where the data-plane verdict allows: member count 0/1). A single node
|
||||
/// still opens — its own redundancy singleton names it Primary in its first snapshot.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Assertions are on what was PUBLISHED, in order, so a missing publish cannot pass for a closed seed.
|
||||
/// </remarks>
|
||||
public sealed class DriverHostActorHistoryUpdateRoleViewTests : RuntimeActorTestBase
|
||||
{
|
||||
private static readonly NodeId TestNode = NodeId.Parse("driver-history-update-test");
|
||||
private static readonly TimeSpan Timeout = PresenceBudget;
|
||||
|
||||
/// <summary>The boot window: a single-driver node (data-plane verdict ALLOW) publishes CLOSED for HistoryUpdate until
|
||||
/// a snapshot arrives; its first snapshot naming it Primary opens it.</summary>
|
||||
[Fact]
|
||||
public void A_single_node_is_closed_until_its_first_snapshot_then_opens()
|
||||
{
|
||||
var view = new RecordingHistoryUpdateRoleView();
|
||||
var host = SpawnHost(view, () => 1);
|
||||
|
||||
AwaitAssert(() => view.Published.ShouldBe([false]), duration: Timeout);
|
||||
|
||||
host.Tell(Snapshot(RedundancyRole.Primary));
|
||||
AwaitAssert(() => view.Published[^1].ShouldBeTrue(), duration: Timeout);
|
||||
}
|
||||
|
||||
/// <summary>Once a snapshot has arrived, an unknown role on a single-driver cluster (a snapshot that omits this
|
||||
/// node) follows the data-plane rule and opens — a real verdict exists, it just does not name us.</summary>
|
||||
[Fact]
|
||||
public void After_a_snapshot_an_unknown_role_follows_the_data_plane_rule()
|
||||
{
|
||||
var view = new RecordingHistoryUpdateRoleView();
|
||||
var host = SpawnHost(view, () => 1);
|
||||
AwaitAssert(() => view.Published.ShouldBe([false]), duration: Timeout);
|
||||
|
||||
host.Tell(new RedundancyStateChanged([], CorrelationId.NewId()));
|
||||
AwaitAssert(() => view.Published[^1].ShouldBeTrue(), duration: Timeout);
|
||||
}
|
||||
|
||||
/// <summary>A Secondary stays closed after its snapshot; becoming Primary opens it.</summary>
|
||||
[Fact]
|
||||
public void A_secondary_stays_closed_and_a_promotion_opens()
|
||||
{
|
||||
var view = new RecordingHistoryUpdateRoleView();
|
||||
var host = SpawnHost(view, () => 2);
|
||||
AwaitAssert(() => view.Published.ShouldBe([false]), duration: Timeout);
|
||||
|
||||
host.Tell(Snapshot(RedundancyRole.Secondary));
|
||||
AwaitAssert(() => view.Published.Count.ShouldBeGreaterThanOrEqualTo(2), duration: Timeout);
|
||||
view.Published.ShouldAllBe(v => !v);
|
||||
|
||||
host.Tell(Snapshot(RedundancyRole.Primary));
|
||||
AwaitAssert(() => view.Published[^1].ShouldBeTrue(), duration: Timeout);
|
||||
}
|
||||
|
||||
/// <summary>The policy itself: no snapshot ⇒ closed whatever the role/count; with one ⇒ the data-plane verdict.</summary>
|
||||
[Theory]
|
||||
[InlineData(false, null, 0, false)]
|
||||
[InlineData(false, null, 1, false)]
|
||||
[InlineData(false, RedundancyRole.Primary, 1, false)]
|
||||
[InlineData(true, null, 1, true)]
|
||||
[InlineData(true, null, 2, false)]
|
||||
[InlineData(true, RedundancyRole.Primary, 2, true)]
|
||||
[InlineData(true, RedundancyRole.Secondary, 2, false)]
|
||||
[InlineData(true, RedundancyRole.Detached, 1, false)]
|
||||
public void Policy_is_the_data_plane_verdict_gated_on_a_snapshot(
|
||||
bool snapshot, RedundancyRole? role, int members, bool expected) =>
|
||||
PrimaryGatePolicy.ShouldServeHistoryUpdate(snapshot, role, members).ShouldBe(expected);
|
||||
|
||||
/// <summary>The production view is seeded CLOSED (unlike the data-plane verdict's allow seed).</summary>
|
||||
[Fact]
|
||||
public void The_role_view_is_seeded_closed()
|
||||
{
|
||||
var view = new RedundancyRoleView();
|
||||
view.ShouldServeHistoryUpdate.ShouldBeFalse();
|
||||
view.ShouldActAsPrimary.ShouldBeTrue("control: the data-plane seed is the single-node allow");
|
||||
view.PublishHistoryUpdateRole(true);
|
||||
view.ShouldServeHistoryUpdate.ShouldBeTrue();
|
||||
}
|
||||
|
||||
private IActorRef SpawnHost(IHistoryUpdateRoleView view, Func<int> driverMemberCount) =>
|
||||
Sys.ActorOf(DriverHostActor.Props(
|
||||
NewInMemoryDbFactory(),
|
||||
TestNode,
|
||||
ackRouter: null,
|
||||
driverMemberCountProvider: driverMemberCount,
|
||||
historyUpdateRoleView: view));
|
||||
|
||||
private static RedundancyStateChanged Snapshot(RedundancyRole role) =>
|
||||
new(
|
||||
[
|
||||
new NodeRedundancyState(TestNode, role,
|
||||
IsClusterLeader: role == RedundancyRole.Primary,
|
||||
IsDriverPrimary: role == RedundancyRole.Primary,
|
||||
AsOfUtc: DateTime.UtcNow),
|
||||
],
|
||||
CorrelationId.NewId());
|
||||
|
||||
/// <summary>Remembers every published verdict, in order.</summary>
|
||||
private sealed class RecordingHistoryUpdateRoleView : IHistoryUpdateRoleView
|
||||
{
|
||||
private readonly List<bool> _published = [];
|
||||
private readonly Lock _gate = new();
|
||||
|
||||
public IReadOnlyList<bool> Published
|
||||
{
|
||||
get { lock (_gate) return _published.ToArray(); }
|
||||
}
|
||||
|
||||
public bool ShouldServeHistoryUpdate
|
||||
{
|
||||
get { lock (_gate) return _published.Count > 0 && _published[^1]; }
|
||||
}
|
||||
|
||||
public void PublishHistoryUpdateRole(bool shouldServeHistoryUpdate)
|
||||
{
|
||||
lock (_gate) _published.Add(shouldServeHistoryUpdate);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user