test(r2-04): primary-gate boot-window guards (deny-on-unknown multi-node, allow single-node)
This commit is contained in:
Joseph Doherty
+425
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// 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 <c>driverMemberCountProvider</c> Props seam.
|
||||
/// </summary>
|
||||
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 ----------------
|
||||
|
||||
/// <summary>Role unknown + a real driver peer (count 2) ⇒ RouteNodeWrite is DENIED with the
|
||||
/// boot-window reason and the driver sees no write.</summary>
|
||||
[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<DriverHostActor.NodeWriteResult>(Timeout);
|
||||
result.Success.ShouldBeFalse();
|
||||
result.Reason.ShouldBe("not primary (role unknown)");
|
||||
recorder.Writes.ShouldBeEmpty();
|
||||
}
|
||||
|
||||
/// <summary>Role unknown + no driver peer (count 1) ⇒ the write is SERVICED (single-node / boot-window
|
||||
/// posture preserved).</summary>
|
||||
[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<DriverHostActor.NodeWriteResult>(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();
|
||||
}
|
||||
|
||||
/// <summary>A Primary snapshot services even at count 2; a Secondary snapshot denies with the steady-state
|
||||
/// reason (distinct from the boot-window reason).</summary>
|
||||
[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<DriverHostActor.NodeWriteResult>(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<DriverHostActor.NodeWriteResult>(Timeout);
|
||||
denied.Success.ShouldBeFalse();
|
||||
denied.Reason.ShouldBe("not primary");
|
||||
}
|
||||
|
||||
/// <summary>The denial meter carries the site + reason tags.</summary>
|
||||
[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<DriverHostActor.NodeWriteResult>(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 ----------------
|
||||
|
||||
/// <summary>Role unknown + count 2 ⇒ a native-alarm ack is dropped at Warning and the denial meter
|
||||
/// increments with site=ack.</summary>
|
||||
[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 ----------------
|
||||
|
||||
/// <summary>Role unknown + count 2 ⇒ ForwardNativeAlarm still writes the (ungated) OPC UA condition update
|
||||
/// but publishes NO cluster-wide alerts transition.</summary>
|
||||
[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<OpcUaPublishActor.AlarmStateUpdate>(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));
|
||||
}
|
||||
|
||||
/// <summary>Role unknown + count 1 ⇒ ForwardNativeAlarm publishes the alerts transition (single-node
|
||||
/// posture preserved).</summary>
|
||||
[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<OpcUaPublishActor.AlarmStateUpdate>(Timeout);
|
||||
alerts.ExpectMsg<AlarmTransitionEvent>(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<OtOpcUaConfigDbContext> 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<string> { "driver" },
|
||||
driverMemberCountProvider: () => driverMemberCount));
|
||||
|
||||
actor.Tell(new DispatchDeployment(dep, RevA, CorrelationId.NewId()));
|
||||
coordinator.ExpectMsg<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
|
||||
return actor;
|
||||
}
|
||||
|
||||
private (IActorRef Actor, TestProbe Publish) SpawnAlarmHost(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext> 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<string> { "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<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
|
||||
publish.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(Timeout);
|
||||
return (actor, publish);
|
||||
}
|
||||
|
||||
private void SubscribeToAlerts(TestProbe probe)
|
||||
{
|
||||
DistributedPubSub.Get(Sys).Mediator.Tell(
|
||||
new Subscribe(ScriptedAlarmHostActor.AlertsTopic, probe.Ref), probe.Ref);
|
||||
probe.ExpectMsg<SubscribeAck>(Timeout);
|
||||
}
|
||||
|
||||
private static DeploymentId SeedWriteTagDeployment(IDbContextFactory<OtOpcUaConfigDbContext> 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<OtOpcUaConfigDbContext> 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<OtOpcUaConfigDbContext> 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<WriteRequest> 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<string> SupportedTypes => new[] { _supportedType };
|
||||
}
|
||||
|
||||
private sealed class RecordingDriver : IDriver, IWritable
|
||||
{
|
||||
private readonly ConcurrentQueue<WriteRequest> _writes = new();
|
||||
public string DriverInstanceId { get; private set; } = string.Empty;
|
||||
public string DriverType { get; private set; } = string.Empty;
|
||||
public IReadOnlyList<WriteRequest> 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<IReadOnlyList<WriteResult>> WriteAsync(
|
||||
IReadOnlyList<WriteRequest> writes, CancellationToken cancellationToken)
|
||||
{
|
||||
foreach (var w in writes) _writes.Enqueue(w);
|
||||
return Task.FromResult<IReadOnlyList<WriteResult>>(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<KeyValuePair<string, object?>[]> _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<long>((_, 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();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user