diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs index 3f6733f74..0f1aa859d 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs @@ -199,6 +199,10 @@ internal static class ProjectedEventDelivery private static readonly Dictionary>> s_registry = new(StringComparer.Ordinal); private static int s_registered; private static int s_sweepAt = 1024; + private static int s_unsharedSinceSweep; + + /// At the cap, refusals between two sweeps for dead shapes. + private const int UnsharedResweepInterval = 256; /// Live distinct filter shapes kept for sharing; past this a new filter gets a group of its own (no sharing). internal const int MaxSharedFilterGroups = 4096; @@ -225,10 +229,9 @@ internal static class ProjectedEventDelivery var handler = s_eventQueue!(item); if (s_filterToUse!(item) is not EventFilter filter) return false; if (handler is null) return true; // Disabled: the SDK queues nothing (no handler) - if (!cache.TryMarkQueued(item)) return true; // the SDK's duplicate check (see EventDeliveryCache) - var locales = item.Session?.PreferredLocales; // string[]?, exactly what the SDK hands its FilterContext and TranslateValue var group = GroupOf(filter, server); + if (!cache.TryMarkQueued(item)) return true; // the SDK's duplicate check (see EventDeliveryCache) var projection = cache.ProjectionFor(group, LocalesKey(locales), () => new EventDeliveryCache.Projection( () => Passes(server, group, locales, e), () => Project(server, group, locales, e))); if (!projection.Passes || handler.SetQueueOverflowIfFull()) return true; @@ -313,9 +316,21 @@ internal static class ProjectedEventDelivery var created = new FilterGroup(f, server); if (s_registered >= MaxSharedFilterGroups) { - OtOpcUaTelemetry.ConditionEventFilterGroupsUnshared.Add(1); - Interlocked.Increment(ref s_unsharedGroups); - return created; + // Full: re-check for dead shapes after every UnsharedResweepInterval refusals, so a registry that filled + // once does not refuse sharing for good, and one that is genuinely full of live shapes is not swept on + // every new filter. + if (++s_unsharedSinceSweep >= UnsharedResweepInterval) + { + s_unsharedSinceSweep = 0; + SweepRegistry(); + } + + if (s_registered >= MaxSharedFilterGroups) + { + OtOpcUaTelemetry.ConditionEventFilterGroupsUnshared.Add(1); + Interlocked.Increment(ref s_unsharedGroups); + return created; + } } s_registered++; @@ -337,8 +352,9 @@ internal static class ProjectedEventDelivery } s_registered = live; - // Amortised: the next sweep only once as many shapes again have been registered (never more often than every 1024). - s_sweepAt = Math.Max(1024, live * 2); + // Amortised: the next sweep only once as many shapes again have been registered (never more often than every 1024), + // and never above the cap — a threshold the count can no longer reach would stop sweeping, and sharing, for good. + s_sweepAt = Math.Min(MaxSharedFilterGroups, Math.Max(1024, live * 2)); } private static long s_unsharedGroups; @@ -361,7 +377,7 @@ internal static class ProjectedEventDelivery var path = new List(); foreach (var name in clause.BrowsePath) path.Add(name.ToString()); - parts.Add(string.Join("/", path) + "@" + clause.AttributeId); + parts.Add(clause.TypeDefinitionId + ":" + string.Join("/", path) + "@" + clause.AttributeId + "[" + clause.IndexRange + "]"); } return string.Join("|", parts);