diff --git a/archreview/plans/R2-04-failure-visibility-plan.md.tasks.json b/archreview/plans/R2-04-failure-visibility-plan.md.tasks.json index 2c5d4a97..20ea2898 100644 --- a/archreview/plans/R2-04-failure-visibility-plan.md.tasks.json +++ b/archreview/plans/R2-04-failure-visibility-plan.md.tasks.json @@ -12,7 +12,7 @@ { "id": "T8", "subject": "Galaxy skip-gated live smoke: normal advised write still returns Good post-change", "status": "completed", "blockedBy": ["T6"] }, { "id": "T9", "subject": "PrimaryGatePolicy pure class + truth-table tests + otopcua.redundancy.primary_gate_denied counter (03/S4)", "status": "completed", "blockedBy": [] }, { "id": "T10", "subject": "Wire DriverHostActor gates (:977/:1039/:1095) through PrimaryGatePolicy with member-count provider + S5 log-once", "status": "completed", "blockedBy": ["T9"] }, - { "id": "T11", "subject": "DriverHostActorPrimaryGateTests: deny-on-unknown multi-node, allow single-node, ack/emit variants, meter tags", "status": "pending", "blockedBy": ["T10"] }, + { "id": "T11", "subject": "DriverHostActorPrimaryGateTests: deny-on-unknown multi-node, allow single-node, ack/emit variants, meter tags", "status": "completed", "blockedBy": ["T10"] }, { "id": "T12", "subject": "ScriptedAlarmHostActor alerts-emit gate adopts PrimaryGatePolicy (consistency extension)", "status": "pending", "blockedBy": ["T9"] }, { "id": "T13", "subject": "In-process 2-node delivery-path test (TwoNodeClusterHarness): DPS-delivered snapshot drives the gate", "status": "pending", "blockedBy": ["T11"] }, { "id": "T14", "subject": "docs/Redundancy.md: gate policy table, boot-window client experience, new meters, Galaxy fail-closed semantics", "status": "pending", "blockedBy": [] }, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryGateTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryGateTests.cs new file mode 100644 index 00000000..036a9007 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryGateTests.cs @@ -0,0 +1,425 @@ +using System.Collections.Concurrent; +using System.Diagnostics.Metrics; +using System.Text.Json; +using Akka.Actor; +using Akka.Cluster.Tools.PublishSubscribe; +using Akka.TestKit; +using Microsoft.EntityFrameworkCore; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Engines; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy; +using ZB.MOM.WW.OtOpcUa.Commons.Observability; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; +using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// archreview 03/S4 — the boot-window primary-gate guard. On a MULTI-driver cluster (member count > 1) +/// while the role is UNKNOWN, the Primary data-plane gates (inbound write, native ack, native alerts emit) +/// default-DENY; on a SINGLE-driver cluster (count <= 1) they default-ALLOW (boot-window / single-node +/// posture). Driver member count is injected via the driverMemberCountProvider Props seam. +/// +public sealed class DriverHostActorPrimaryGateTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-pgate-test"); + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); + private static readonly DateTime Ts = new(2026, 7, 13, 10, 0, 0, DateTimeKind.Utc); + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5); + + // ---------------- write gate ---------------- + + /// Role unknown + a real driver peer (count 2) ⇒ RouteNodeWrite is DENIED with the + /// boot-window reason and the driver sees no write. + [Fact] + public void Unknown_role_multi_driver_denies_write_with_role_unknown_reason() + { + var db = NewInMemoryDbFactory(); + var recorder = new RecordingDriverFactory("Modbus"); + var dep = SeedWriteTagDeployment(db); + var actor = SpawnWriteHost(db, dep, recorder, driverMemberCount: 2); + + var asker = CreateTestProbe(); + actor.Tell(new DriverHostActor.RouteNodeWrite("eq-1/speed", 123.0), asker.Ref); + + var result = asker.ExpectMsg(Timeout); + result.Success.ShouldBeFalse(); + result.Reason.ShouldBe("not primary (role unknown)"); + recorder.Writes.ShouldBeEmpty(); + } + + /// Role unknown + no driver peer (count 1) ⇒ the write is SERVICED (single-node / boot-window + /// posture preserved). + [Fact] + public void Unknown_role_single_driver_services_write() + { + var db = NewInMemoryDbFactory(); + var recorder = new RecordingDriverFactory("Modbus"); + var dep = SeedWriteTagDeployment(db); + var actor = SpawnWriteHost(db, dep, recorder, driverMemberCount: 1); + + // The gate must ALLOW (count 1, role unknown). The driver child's async connect can lag the ApplyAck, + // so a first write may transiently report "driver not connected" — retry until connected (a real + // client retries too). The point under test is that the gate never denies (never "not primary"). + AwaitAssert(() => + { + var asker = CreateTestProbe(); + actor.Tell(new DriverHostActor.RouteNodeWrite("eq-1/speed", 123.0), asker.Ref); + var res = asker.ExpectMsg(Timeout); + res.Reason.ShouldNotBe("not primary"); + res.Reason.ShouldNotBe("not primary (role unknown)"); + res.Success.ShouldBeTrue(res.Reason); + }, duration: TimeSpan.FromSeconds(10)); + recorder.Writes.ShouldNotBeEmpty(); + } + + /// A Primary snapshot services even at count 2; a Secondary snapshot denies with the steady-state + /// reason (distinct from the boot-window reason). + [Fact] + public void Known_role_wins_over_member_count() + { + var db = NewInMemoryDbFactory(); + var recorder = new RecordingDriverFactory("Modbus"); + var dep = SeedWriteTagDeployment(db); + var actor = SpawnWriteHost(db, dep, recorder, driverMemberCount: 2); + + // Primary snapshot ⇒ serviced even with a driver peer. Retry through the transient driver-connect + // window; the invariant under test is that the gate never denies a Primary. + TellRole(actor, RedundancyRole.Primary); + AwaitAssert(() => + { + var asker1 = CreateTestProbe(); + actor.Tell(new DriverHostActor.RouteNodeWrite("eq-1/speed", 1.0), asker1.Ref); + var res = asker1.ExpectMsg(Timeout); + res.Reason.ShouldNotBe("not primary"); + res.Success.ShouldBeTrue(res.Reason); + }, duration: TimeSpan.FromSeconds(10)); + + // Secondary snapshot ⇒ denied with the steady-state reason. + TellRole(actor, RedundancyRole.Secondary); + var asker2 = CreateTestProbe(); + actor.Tell(new DriverHostActor.RouteNodeWrite("eq-1/speed", 2.0), asker2.Ref); + var denied = asker2.ExpectMsg(Timeout); + denied.Success.ShouldBeFalse(); + denied.Reason.ShouldBe("not primary"); + } + + /// The denial meter carries the site + reason tags. + [Fact] + public void Denied_write_increments_primary_gate_denied_meter_with_tags() + { + using var recorder = new MeterRecorder("otopcua.redundancy.primary_gate_denied"); + var db = NewInMemoryDbFactory(); + var factory = new RecordingDriverFactory("Modbus"); + var dep = SeedWriteTagDeployment(db); + var actor = SpawnWriteHost(db, dep, factory, driverMemberCount: 2); + + var asker = CreateTestProbe(); + actor.Tell(new DriverHostActor.RouteNodeWrite("eq-1/speed", 9.0), asker.Ref); + asker.ExpectMsg(Timeout).Success.ShouldBeFalse(); + + AwaitAssert(() => + { + recorder.Total.ShouldBeGreaterThanOrEqualTo(1); + recorder.WithTag("site", "write").ShouldBeGreaterThanOrEqualTo(1); + recorder.WithTag("reason", "role-unknown").ShouldBeGreaterThanOrEqualTo(1); + }, duration: Timeout); + } + + // ---------------- ack gate ---------------- + + /// Role unknown + count 2 ⇒ a native-alarm ack is dropped at Warning and the denial meter + /// increments with site=ack. + [Fact] + public void Unknown_role_multi_driver_drops_native_ack_with_warning_and_meter() + { + using var meter = new MeterRecorder("otopcua.redundancy.primary_gate_denied"); + var db = NewInMemoryDbFactory(); + var dep = SeedAlarmTagDeployment(db); + var (actor, _) = SpawnAlarmHost(db, dep, driverMemberCount: 2); + + EventFilter.Warning(contains: "role unknown").ExpectOne(() => + actor.Tell(new DriverHostActor.RouteNativeAlarmAck("eq-1/temp_hi", Comment: null, OperatorUser: "op"))); + + AwaitAssert(() => + { + meter.WithTag("site", "ack").ShouldBeGreaterThanOrEqualTo(1); + }, duration: Timeout); + } + + // ---------------- alarm-emit gate ---------------- + + /// Role unknown + count 2 ⇒ ForwardNativeAlarm still writes the (ungated) OPC UA condition update + /// but publishes NO cluster-wide alerts transition. + [Fact] + public void Unknown_role_multi_driver_suppresses_alerts_emit_but_updates_condition() + { + var db = NewInMemoryDbFactory(); + var dep = SeedAlarmTagDeployment(db); + var alerts = CreateTestProbe(); + SubscribeToAlerts(alerts); + var (actor, publish) = SpawnAlarmHost(db, dep, driverMemberCount: 2); + + actor.Tell(RaiseAlarm()); + + // The ungated OPC UA condition write still arrives. + publish.ExpectMsg(Timeout).AlarmNodeId.ShouldBe("eq-1/temp_hi"); + // The cluster-wide alerts publish is suppressed (a real Primary peer publishes the single copy). + alerts.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + } + + /// Role unknown + count 1 ⇒ ForwardNativeAlarm publishes the alerts transition (single-node + /// posture preserved). + [Fact] + public void Unknown_role_single_driver_publishes_alerts_emit() + { + var db = NewInMemoryDbFactory(); + var dep = SeedAlarmTagDeployment(db); + var alerts = CreateTestProbe(); + SubscribeToAlerts(alerts); + var (actor, publish) = SpawnAlarmHost(db, dep, driverMemberCount: 1); + + actor.Tell(RaiseAlarm()); + + publish.ExpectMsg(Timeout); + alerts.ExpectMsg(Timeout).AlarmId.ShouldBe("eq-1/temp_hi"); + } + + // ---------------- helpers ---------------- + + private static DriverInstanceActor.AttributeAlarmPublished RaiseAlarm() => + new("drv-1", new AlarmEventArgs( + new StubAlarmHandle(), + SourceNodeId: "Temp", + ConditionId: "Temp.HiHi", + AlarmType: "OffNormalAlarm", + Message: "temperature high", + Severity: AlarmSeverity.High, + SourceTimestampUtc: Ts, + Kind: AlarmTransitionKind.Raise)); + + private static void TellRole(IActorRef host, RedundancyRole role) => + host.Tell(new RedundancyStateChanged( + new[] + { + new NodeRedundancyState(TestNode, role, + IsClusterLeader: role == RedundancyRole.Primary, + IsRoleLeaderForDriver: role == RedundancyRole.Primary, + AsOfUtc: DateTime.UtcNow), + }, + CorrelationId.NewId())); + + private IActorRef SpawnWriteHost( + IDbContextFactory db, DeploymentId dep, IDriverFactory factory, int driverMemberCount) + { + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + driverFactory: factory, + localRoles: new HashSet { "driver" }, + driverMemberCountProvider: () => driverMemberCount)); + + actor.Tell(new DispatchDeployment(dep, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied); + return actor; + } + + private (IActorRef Actor, TestProbe Publish) SpawnAlarmHost( + IDbContextFactory db, DeploymentId dep, int driverMemberCount) + { + var coordinator = CreateTestProbe(); + var publish = CreateTestProbe(); + var mux = CreateTestProbe(); + var vtHost = CreateTestProbe(); + + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + dependencyMux: mux.Ref, + opcUaPublishActor: publish.Ref, + virtualTagEvaluator: NullVirtualTagEvaluator.Instance, + virtualTagHostOverride: vtHost.Ref, + driverMemberCountProvider: () => driverMemberCount)); + + actor.Tell(new DispatchDeployment(dep, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied); + publish.ExpectMsg(Timeout); + return (actor, publish); + } + + private void SubscribeToAlerts(TestProbe probe) + { + DistributedPubSub.Get(Sys).Mediator.Tell( + new Subscribe(ScriptedAlarmHostActor.AlertsTopic, probe.Ref), probe.Ref); + probe.ExpectMsg(Timeout); + } + + private static DeploymentId SeedWriteTagDeployment(IDbContextFactory db) + { + var artifact = JsonSerializer.SerializeToUtf8Bytes(new + { + Namespaces = new[] { new { NamespaceId = "ns-eq", Kind = 0 } }, + DriverInstances = new[] + { + new + { + DriverInstanceRowId = Guid.NewGuid(), + DriverInstanceId = "drv-1", + Name = "drv-1", + DriverType = "Modbus", + Enabled = true, + DriverConfig = "{}", + NamespaceId = "ns-eq", + }, + }, + Tags = new[] + { + new + { + TagId = "tag-0", + EquipmentId = "eq-1", + DriverInstanceId = "drv-1", + Name = "speed", + FolderPath = (string?)null, + DataType = "Double", + TagConfig = JsonSerializer.Serialize(new { FullName = "40001" }), + }, + }, + }); + return SealArtifact(db, artifact); + } + + private static DeploymentId SeedAlarmTagDeployment(IDbContextFactory db) + { + var artifact = JsonSerializer.SerializeToUtf8Bytes(new + { + Namespaces = new[] { new { NamespaceId = "ns-eq", Kind = 0 } }, + DriverInstances = new[] { new { DriverInstanceId = "drv-1", NamespaceId = "ns-eq" } }, + Tags = new[] + { + new + { + TagId = "tag-0", + EquipmentId = "eq-1", + DriverInstanceId = "drv-1", + Name = "temp_hi", + FolderPath = (string?)null, + DataType = "Boolean", + TagConfig = JsonSerializer.Serialize(new + { + FullName = "Temp.HiHi", + alarm = new { alarmType = "OffNormalAlarm", severity = 700 }, + }), + }, + }, + }); + return SealArtifact(db, artifact); + } + + private static DeploymentId SealArtifact(IDbContextFactory db, byte[] artifact) + { + var id = DeploymentId.NewId(); + using var ctx = db.CreateDbContext(); + ctx.Deployments.Add(new Deployment + { + DeploymentId = id.Value, + RevisionHash = RevA.Value, + Status = DeploymentStatus.Sealed, + CreatedBy = "test", + SealedAtUtc = DateTime.UtcNow, + ArtifactBlob = artifact, + }); + ctx.SaveChanges(); + return id; + } + + private sealed class StubAlarmHandle : IAlarmSubscriptionHandle + { + public string DiagnosticId => "stub-alarm-sub"; + } + + private sealed class RecordingDriverFactory : IDriverFactory + { + private readonly string _supportedType; + private readonly RecordingDriver _driver = new(); + public RecordingDriverFactory(string supportedType) { _supportedType = supportedType; } + public IReadOnlyList Writes => _driver.Writes; + + public IDriver? TryCreate(string driverType, string driverInstanceId, string driverConfigJson) + { + if (!string.Equals(driverType, _supportedType, StringComparison.Ordinal)) return null; + _driver.Bind(driverInstanceId, driverType); + return _driver; + } + + public IReadOnlyCollection SupportedTypes => new[] { _supportedType }; + } + + private sealed class RecordingDriver : IDriver, IWritable + { + private readonly ConcurrentQueue _writes = new(); + public string DriverInstanceId { get; private set; } = string.Empty; + public string DriverType { get; private set; } = string.Empty; + public IReadOnlyList Writes => _writes.ToArray(); + public void Bind(string id, string type) { DriverInstanceId = id; DriverType = type; } + public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; + public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; + public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask; + public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, LastError: null); + public long GetMemoryFootprint() => 0; + public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask; + public Task> WriteAsync( + IReadOnlyList writes, CancellationToken cancellationToken) + { + foreach (var w in writes) _writes.Enqueue(w); + return Task.FromResult>(writes.Select(_ => new WriteResult(0u)).ToArray()); + } + } + + private sealed class MeterRecorder : IDisposable + { + private readonly string _name; + private readonly MeterListener _listener; + private long _total; + private readonly List[]> _tagSets = new(); + private readonly object _gate = new(); + + public MeterRecorder(string instrumentName) + { + _name = instrumentName; + _listener = new MeterListener + { + InstrumentPublished = (instrument, listener) => + { + if (instrument.Meter.Name == OtOpcUaTelemetry.MeterName && instrument.Name == _name) + listener.EnableMeasurementEvents(instrument); + } + }; + _listener.SetMeasurementEventCallback((_, value, tags, _) => + { + lock (_gate) { _total += value; _tagSets.Add(tags.ToArray()); } + }); + _listener.Start(); + } + + public long Total { get { lock (_gate) return _total; } } + + public int WithTag(string key, string value) + { + lock (_gate) return _tagSets.Count(set => set.Any(t => t.Key == key && Equals(t.Value, value))); + } + + public void Dispose() => _listener.Dispose(); + } +}