diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs index 877cc57f1..5ef544255 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs @@ -33,8 +33,11 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; /// /// Duplicate check. The SDK compares the event INSTANCE against the last 1000 queued handles. A condition /// snapshot reaches an item only through this per-report delivery, so "already queued in this report" is the same -/// question, asked of a set instead of the queue (it stays true after the event leaves the queue, where the SDK's -/// would not — a second delivery of one report to one item never happens on purpose). +/// question, asked of a set instead of the queue. It matters: a walk can reach one notifier twice (a Line folder above +/// two equipment folders the alarm is wired into), and the SDK's scan suppressed that second copy too. The set +/// stays true after the event has left the queue, where the SDK's scan would not — a second visit within one walk +/// happens long before anything is published; at a full DiscardOldest=false queue the SDK also counted that second +/// visit as one more discarded event, this counts it once. /// /// /// Fallback. Anything outside that shape — the private members not found (an SDK upgrade; pinned by a guard @@ -275,7 +278,9 @@ internal static class ProjectedEventDelivery private static FilterGroup GroupOf(EventFilter filter, IServerInternal server) => s_groups.GetValue(filter, f => { - var key = SelectKey(f); + // The where clause takes part in the bucket (its hash is value-based), so filters that share a select list but + // differ in their where clauses (per-source filters) do not all land in one bucket scanned under Lock. + var key = SelectKey(f) + "#" + f.WhereClause?.GetHashCode().ToString(System.Globalization.CultureInfo.InvariantCulture); lock (s_registryLock) { if (s_registry.TryGetValue(key, out var bucket)) @@ -325,9 +330,14 @@ internal static class ProjectedEventDelivery } /// An unambiguous key for a locale list (length-prefixed entries; null distinct from empty). - private static string LocalesKey(string[]? locales) + private static readonly ConditionalWeakTable s_localeKeys = new(); + + /// An unambiguous key for a locale list, cached per array (a session's array is fixed for its lifetime). + internal static string LocalesKey(string[]? locales) => + locales is null ? "null" : s_localeKeys.GetValue(locales, static l => BuildLocalesKey(l)); + + private static string BuildLocalesKey(string[] locales) { - if (locales is null) return "null"; var key = new System.Text.StringBuilder().Append(locales.Length).Append(':'); foreach (var locale in locales) key.Append(locale?.Length ?? -1).Append(':').Append(locale).Append(';'); return key.ToString(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs index 8cc860272..617539a6f 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs @@ -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(); } + /// + /// 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. + /// + [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")); + } + + /// + /// 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 must take part in + /// that exactly like the SDK's own (each on a fresh server). + /// + [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> 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"); + } + + /// + /// 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). + /// + [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); + /// /// The projected path rests on PRIVATE members of the pinned SDK's MonitoredItem (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