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

Merge feat/674-historyupdate-posture follow-ups: HistoryUpdate closed until first redundancy snapshot, node-level v1 refusal, host wiring tests, plan-before-gate, UTC skew (#674 review)

Browse Source
This commit is contained in:
Joseph Doherty
2026-10-11 11:41:36 -04:00
parent 7594120a55 232c03ed1b
commit 029a916288
16 changed files with 618 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/StatusCodeOracleTests.cs
-5
View File
@@ -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()
{
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SchemaV1PostureTests.cs
+11 -11
View File
@@ -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]);
tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/OpcUa/OtOpcUaServerHostedServiceHistoryUpdateTests.cs
+116
View File
@@ -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);
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/History/HistoryUpdateTests.cs
+120 -1
View File
@@ -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();
}
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorHistoryUpdateRoleViewTests.cs
+133
View File
@@ -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);
}
}
}