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

fix(opcua): #855 perf review 2 — where clause in the group bucket key; pin dedupe, modify and keys

Browse Source 
- Filter-group buckets include the where clause's (value-based) hash, so
  per-source filters sharing a select list are not all scanned under Lock.
- Locale keys are cached per session array.
- Tests: a notifier reached twice in one walk (Line above two wired
  equipment) gets each event once on both paths (fails without the
  per-report set); queued events re-projected by a modify are identical on
  both paths; locale keys unambiguous and where clauses hash by value.

Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
Joseph Doherty
2026-11-11 14:10:02 -05:00
parent ca0e73fbca
commit b77bb2ded1
2 changed files with 133 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs
+118
View File
@@ -5,6 +5,7 @@ using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
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.Core.Authorization;
using MonitoredItem = Opc.Ua.Client.MonitoredItem;
@@ -180,6 +181,123 @@ public sealed class ProjectedEventDeliveryWireTests
.ToList();
}
/// <summary>
/// A walk that reaches one notifier twice — a Line folder above two equipment folders the native alarm is wired into
/// (alarm → E1 → Line, alarm → E2 → Line) — must queue each event to the Line's item once, as the SDK's duplicate
/// scan did. Both paths, compared; no EventId may repeat.
/// </summary>
[Fact]
public async Task A_notifier_reached_twice_in_one_walk_gets_each_event_once_on_both_paths()
{
var ct = TestContext.Current.CancellationToken;
await using var rig = await ServerEventRig.StartAsync("Projected855Twice", encrypted: false, ct);
rig.KeyRoles = [];
rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
var sink = rig.Sink;
sink.EnsureFolder(ServerEventRig.RawDeviceFolder, parentNodeId: null, displayName: "dev1", realm: AddressSpaceRealm.Raw);
sink.MaterialiseAlarmCondition(ServerEventRig.RawAlarmPath, ServerEventRig.RawDeviceFolder, "temp_hi", "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: true);
sink.EnsureFolder("L1", parentNodeId: null, displayName: "L1", realm: AddressSpaceRealm.Uns);
sink.EnsureFolder("E1", parentNodeId: "L1", displayName: "E1", realm: AddressSpaceRealm.Uns);
sink.EnsureFolder("E2", parentNodeId: "L1", displayName: "E2", realm: AddressSpaceRealm.Uns);
// A condition directly under L1 makes L1 an event notifier (only folders a condition hangs under, or is wired into, are).
sink.MaterialiseAlarmCondition("line-alarm", "L1", "line", "AlarmCondition", 500, AddressSpaceRealm.Uns, isNative: false);
sink.WireAlarmNotifiers(ServerEventRig.RawAlarmPath, AddressSpaceRealm.Raw, ["E1", "E2"], AddressSpaceRealm.Uns);
using var session = await rig.ConnectAsync(anonymous: false, ct);
var events = new RecordedEvents();
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
var onLine = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, "L1"));
var onE1 = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, "E1"));
await subscription.ApplyChangesAsync(ct);
onLine.Status.Error.ShouldBeNull();
foreach (var (prefix, projected) in new[] { ("p", true), ("s", false) })
{
rig.Nm.UseProjectedEventDelivery = projected;
for (var i = 0; i < 5; i++) rig.Project(native: true, State($"{prefix}{i}"));
await RecordedEvents.WaitUntilAsync(() => events.Has(onLine, $"{prefix}4") && events.Has(onE1, $"{prefix}4"));
await Task.Delay(TimeSpan.FromMilliseconds(300), ct);
rig.Nm.UseProjectedEventDelivery = true;
}
var ids = events.Of(onLine)
.Where(f => f[RecordedEvents.AckedField].TryGetValue(out bool _))
.Select(f => Convert.ToHexString(f[RecordedEvents.EventIdField].GetByteString().ToArray()!))
.ToList();
ids.Count.ShouldBe(10);
ids.Distinct().Count().ShouldBe(ids.Count, "each event once per item, though the walk reaches L1 twice");
Normalised(events.Of(onLine), "p").ShouldBe(Normalised(events.Of(onLine), "s"));
}
/// <summary>
/// Events still QUEUED when the item is modified (publishing off, then a new select list) are re-projected by the
/// SDK from each queued event's Handle; the projected path's per-item <see cref="EventFieldList"/> must take part in
/// that exactly like the SDK's own (each on a fresh server).
/// </summary>
[Fact]
public async Task Queued_events_re_projected_by_a_modify_are_identical_on_both_paths()
{
var projected = await ModifyScenarioAsync(projectedPath: true);
var sdk = await ModifyScenarioAsync(projectedPath: false);
projected.ShouldBe(sdk);
projected.Count.ShouldBe(Severities.Length);
}
private static async Task<List<string>> ModifyScenarioAsync(bool projectedPath)
{
var ct = TestContext.Current.CancellationToken;
await using var rig = await ServerEventRig.StartAsync($"Projected855Modify{projectedPath}", encrypted: false, ct);
rig.KeyRoles = [];
rig.MaterialiseNative();
rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
rig.Nm.UseProjectedEventDelivery = projectedPath;
using var session = await rig.ConnectAsync(anonymous: false, ct);
var events = new RecordedEvents();
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
var item = AddItem(subscription, ServerEventRig.UnsNodeId(session, EqA), OfTypeAlarm(alternative: false));
await subscription.ApplyChangesAsync(ct);
item.Status.Error.ShouldBeNull();
await subscription.SetPublishingModeAsync(false, ct);
for (var i = 0; i < Severities.Length; i++)
{
rig.Project(native: true, new AlarmConditionSnapshot(
true, false, false, true, AlarmShelvingKind.Unshelved, Severities[i], $"m{i}"));
}
item.Filter = OfTypeAlarm(alternative: true); // a new select list while every event is still queued
await subscription.ApplyChangesAsync(ct);
item.Status.Error.ShouldBeNull();
await subscription.SetPublishingModeAsync(true, ct);
await RecordedEvents.WaitUntilAsync(() => events.Has(item, $"m{Severities.Length - 1}"));
await Task.Delay(TimeSpan.FromMilliseconds(300), ct);
return Normalised(events.Of(item), "m");
}
/// <summary>
/// The two keys the grouping rests on: the locale key is unambiguous (null, empty, one empty locale, and a locale
/// containing the separator are all distinct), and a where clause hashes by value (two value-equal filters land in
/// one bucket and share a projection).
/// </summary>
[Fact]
public void Locale_keys_are_unambiguous_and_where_clauses_hash_by_value()
{
string?[] keys =
[
ProjectedEventDelivery.LocalesKey(null), ProjectedEventDelivery.LocalesKey([]),
ProjectedEventDelivery.LocalesKey([""]), ProjectedEventDelivery.LocalesKey(["a\u0001b"]),
ProjectedEventDelivery.LocalesKey(["a", "b"]), ProjectedEventDelivery.LocalesKey(["a;b"]),
ProjectedEventDelivery.LocalesKey(["a", ";b"]),
];
keys.Distinct().Count().ShouldBe(keys.Length);
Narrow(800).WhereClause.GetHashCode().ShouldBe(Narrow(800).WhereClause.GetHashCode());
Narrow(800).WhereClause.GetHashCode().ShouldNotBe(Narrow(900).WhereClause.GetHashCode());
}
private static AlarmConditionSnapshot State(string message) =>
new(true, false, false, true, AlarmShelvingKind.Unshelved, 700, message);
/// <summary>
/// The projected path rests on PRIVATE members of the pinned SDK's <c>MonitoredItem</c> (its lock, its event queue
/// handler, the filter in use). If an SDK upgrade drops one, delivery silently falls back to the SDK's per-item