diff --git a/.claude/rules/data-plane-auth.md b/.claude/rules/data-plane-auth.md index ded979d62..56006accb 100644 --- a/.claude/rules/data-plane-auth.md +++ b/.claude/rules/data-plane-auth.md @@ -217,7 +217,7 @@ Two things to know before relying on any of this: replay off `Lock` one event per hold of a per-item gate, and withholds older deferred events from THAT item only (re-sent condition, or Retain=true of one omitted because NOT retained — never one omitted by ACL). Item state is a `ConditionalWeakTable` keyed by the item; cuts clear when the deferral count hits 0. Tests: - `ConditionRefreshDeferralWireTests`. ⚠️ **Since #855 folder items are delivered per item too:** at every notifier we own, `DeliverConditionEventAsync` serves the `MonitoredNode2`'s attached items itself (`DeliverToNotifierItems`: audit guard → `INodeManager3.ValidateEventRolePermissions` per event, UNCACHED → item gate/cut → `QueueEvent`), never through the SDK channel — so no notifier of ours defers, a flood under `Lock` defers nothing, and folder refreshes (`RefreshDirectItems`, flag on AND off) share the #854 cut. The deferral queue + cut remain as the safety net; an event whose walk suspends is marked deferred BEFORE suspending. Tests produce deferral with `NotifierStall` (the `NotifierDeliveryStall` probe) — holding a folder item's SDK lock now blocks the reporter. `FolderConditionRefreshWireTests`; `docs/plans/2026-11-11-folder-event-per-item-delivery.md`. Folder events follow Part 3's shape — + `ConditionRefreshDeferralWireTests`. ⚠️ **Since #855 folder items are delivered per item too:** at every notifier we own, `DeliverConditionEventAsync` serves the `MonitoredNode2`'s attached items itself (`DeliverToNotifierItems`: audit guard → `INodeManager3.ValidateEventRolePermissions` per event, UNCACHED → item gate/cut → `QueueEvent`), never through the SDK channel — so no notifier of ours defers, a flood under `Lock` defers nothing, and folder refreshes (`RefreshDirectItems`, flag on AND off) share the #854 cut. The deferral queue + cut remain as the safety net; an event whose walk suspends is marked deferred BEFORE suspending. Tests produce deferral with `NotifierStall` (the `NotifierDeliveryStall` probe) — holding a folder item's SDK lock now blocks the reporter. `FolderConditionRefreshWireTests`; `docs/plans/2026-11-11-folder-event-per-item-delivery.md`. Because that queueing runs under `Lock`, event item queues are `RingEventMonitoredItemQueue` (#870, supplied by `OtOpcUaSdkServer.CreateMonitoredItemQueueFactory` via `RingEventQueueFactory`; data-change queues stay the SDK's) — the SDK queue's discard-oldest was a `List.RemoveAt(0)` per stalled item per report. It is storage only (overflow events + DiscardOldest live in the SDK's unchanged `EventQueueHandler`); parity pinned by `RingEventMonitoredItemQueueParityTests` — re-check them on an SDK bump. Folder events follow Part 3's shape — `ReceiveEvents` on EventType + SourceNode (SDK check, kept first) AND our decision; non-condition events on a folder item fail closed under enforcement. Tests: `FolderEventFloodWireTests`. Overflow WITH the ACL doing work (#838 item 2) — denied condition interleaved, folder + `i=2253` items, non-publishing client, concurrent Read, grant revoked diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs index 3af2c41a0..82c1c3d83 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaSdkServer.cs @@ -314,6 +314,27 @@ public sealed class OtOpcUaSdkServer : StandardServer _otOpcUaNodeManager!.InstallApplicationAuditPermissions(auditType); } + /// + /// Test/measurement switch (Gitea #870): false makes servers started afterwards use the SDK's own event queues, so + /// the perf harness can A/B the two in one process. Never cleared in production. + /// + internal static bool RingEventQueues { get; set; } = true; + + /// + /// The SDK's factory with event queues swapped for (Gitea #870: since #855 + /// folder event items are queued under Lock, and the SDK queue's discard-oldest is an O(QueueSize) shift). + /// A factory set through is used as-is. + /// + /// + protected override IMonitoredItemQueueFactory? CreateMonitoredItemQueueFactory( + IServerInternal server, ApplicationConfiguration configuration) + { + var factory = base.CreateMonitoredItemQueueFactory(server, configuration); + if (factory is null || ReferenceEquals(factory, MonitoredItemQueueFactory) || !RingEventQueues) return factory; + // Server-owned (OwnsMonitoredItemQueueFactory: not the property's instance), so disposed on stop with the SDK's. + return new RingEventQueueFactory(factory); + } + /// protected override MasterNodeManager CreateMasterNodeManager( IServerInternal server, ApplicationConfiguration configuration) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/RingEventMonitoredItemQueue.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/RingEventMonitoredItemQueue.cs new file mode 100644 index 000000000..f0f9dec5c --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/RingEventMonitoredItemQueue.cs @@ -0,0 +1,182 @@ +using Opc.Ua; +using Opc.Ua.Server; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; + +/// +/// A ring-buffer replacement for the SDK's (Gitea #870): the same observable +/// behaviour, but discarding the oldest event at a full queue is O(1) instead of a List.RemoveAt(0) shift of +/// the whole queue. +/// +/// +/// +/// Why. Since #855 condition events are queued to folder event items by the reporter, under the node +/// manager's Lock. At a full queue with DiscardOldest the SDK's dequeues the +/// head and enqueues the new event; the SDK queue's dequeue is List.RemoveAt(0), ~1 µs per item at +/// QueueSize 1000 and ~10 µs at 10 000 — paid per stalled item per report while Lock is held. +/// +/// +/// What is reproduced (SDK 2.0.1, decompiled). The SDK queue is pure storage — DiscardOldest, the +/// EventQueueOverflowEventType positions and the discarded/overflow counters all live in +/// , which is unchanged and drives this queue exactly as it drives the SDK's. +/// Reproduced here: starts at 0; at QueueSize 0 throws +/// BadInternalError, and at Count == QueueSize drops the oldest +/// entry first; is FIFO and returns false (value null) when empty; +/// keeps the newest queueSize entries when discardOldest, else the +/// oldest; checks the newest ≤ 1000 entries with the SDK's own +/// (null entries skipped); is +/// false; does nothing observable (the SDK's does nothing at all). A durable queue is +/// never created here — hands that request to the SDK factory, which refuses it +/// exactly as before. +/// +/// +/// Not thread-safe, like the SDK's: is only used under its monitored item's lock. +/// Parity is pinned by RingEventMonitoredItemQueueParityTests. +/// +/// +public sealed class RingEventMonitoredItemQueue : IEventMonitoredItemQueue +{ + /// The SDK's kMaxNoOfEntriesCheckedForDuplicateEvents. + internal const int MaxEntriesCheckedForDuplicates = 1000; + + private const int MinimumCapacity = 4; + + private EventFieldList?[] _buffer = []; + private int _head; + private int _count; + + /// Creates an empty queue (QueueSize 0, as the SDK's). + /// The monitored item the queue belongs to. + public RingEventMonitoredItemQueue(uint monitoredItemId) => MonitoredItemId = monitoredItemId; + + /// + public uint MonitoredItemId { get; } + + /// + public bool IsDurable => false; + + /// + public uint QueueSize { get; private set; } + + /// + public int ItemsInQueue => _count; + + /// Allocated slots (test observation: the buffer grows with use and shrinks with the queue size). + internal int Capacity => _buffer.Length; + + /// + public void SetQueueSize(uint queueSize, bool discardOldest) + { + QueueSize = queueSize; + if (_count > queueSize) + { + var excess = _count - (int)queueSize; + if (discardOldest) + { + for (var i = 0; i < excess; i++) DropHead(); + } + else + { + for (var i = 0; i < excess; i++) + { + _buffer[Slot(_count - 1)] = null; + _count--; + } + } + } + + // Not observable: release a buffer much larger than the new size (a List would keep it). + if (_buffer.Length > MinimumCapacity && _buffer.Length > 2 * Math.Max(queueSize, (uint)MinimumCapacity)) + Resize(Math.Max(_count, MinimumCapacity)); + } + + /// + public bool IsEventContainedInQueue(IFilterTarget instance) + { + var oldest = _count > MaxEntriesCheckedForDuplicates ? _count - MaxEntriesCheckedForDuplicates : 0; + for (var i = _count - 1; i >= oldest; i--) + { + var queued = _buffer[Slot(i)]; + if (queued is not null && EventMonitoredItemQueue.IsEventRaisedFor(queued, instance)) return true; + } + + return false; + } + + /// + public void Enqueue(EventFieldList value) + { + if (QueueSize == 0) + throw new ServiceResultException(StatusCodes.BadInternalError, "Error queueing Event. Queue size is set to 0"); + + if (_count == QueueSize) DropHead(); + + if (_count == _buffer.Length) + Resize((int)Math.Min(QueueSize, (ulong)Math.Max(MinimumCapacity, _buffer.Length * 2L))); + + _buffer[Slot(_count)] = value; + _count++; + } + + /// + public bool Dequeue(out EventFieldList value) + { + value = null!; + if (_count == 0) return false; + value = _buffer[_head]!; + DropHead(); + return true; + } + + /// + public void Dispose() + { + // The SDK's queue disposes nothing; the queue stays usable, as the SDK's does. + } + + private int Slot(int index) + { + var slot = _head + index; + return slot >= _buffer.Length ? slot - _buffer.Length : slot; + } + + private void DropHead() + { + _buffer[_head] = null; + _head = _head + 1 == _buffer.Length ? 0 : _head + 1; + _count--; + if (_count == 0) _head = 0; + } + + private void Resize(int capacity) + { + var next = new EventFieldList?[Math.Max(capacity, _count)]; + for (var i = 0; i < _count; i++) next[i] = _buffer[Slot(i)]; + _buffer = next; + _head = 0; + } +} + +/// +/// The server's (Gitea #870): event queues are +/// ; data-change queues, durable requests and disposal go to the SDK's own +/// factory unchanged. +/// +public sealed class RingEventQueueFactory(IMonitoredItemQueueFactory inner) : IMonitoredItemQueueFactory +{ + private readonly IMonitoredItemQueueFactory _inner = inner ?? throw new ArgumentNullException(nameof(inner)); + + /// + public bool SupportsDurableQueues => _inner.SupportsDurableQueues; + + /// + public IDataChangeMonitoredItemQueue CreateDataChangeQueue(bool isDurable, uint monitoredItemId) => + _inner.CreateDataChangeQueue(isDurable, monitoredItemId); + + /// + public IEventMonitoredItemQueue CreateEventQueue(bool isDurable, uint monitoredItemId) => + isDurable ? _inner.CreateEventQueue(isDurable, monitoredItemId) : new RingEventMonitoredItemQueue(monitoredItemId); + + /// + public void Dispose() => _inner.Dispose(); +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventDeliveryPerfTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventDeliveryPerfTests.cs index 25b14788e..b81ac7f0a 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventDeliveryPerfTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventDeliveryPerfTests.cs @@ -35,6 +35,9 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; /// compiles against the pre-#855 tree for a before/after comparison. /// [Trait("Category", "Perf")] +// Shares a collection with the other user of OtOpcUaSdkServer.RingEventQueues (a process-wide switch the perf +// harness flips), so the wiring assertion never observes it switched off (#870 review). +[Collection(RingEventQueueSwitchCollection.Name)] public sealed class FolderEventDeliveryPerfTests { private const string OptInVariable = "OTOPCUA_EVENT_PERF"; @@ -100,8 +103,40 @@ public sealed class FolderEventDeliveryPerfTests public Task Stalled_items_queue_size_sensitivity(uint queueSize) => RunAsync(50, acl: false, stalled: true, serverObjectProbe: false, queueSize); + /// + /// Gitea #870 A/B in one process (same machine load): N stalled folder items at a full queue, with the server's + /// ring-buffer event queues vs the SDK's own EventMonitoredItemQueue (List.RemoveAt(0) per discard). + /// Each round runs both arms, alternating which goes first, so a load spike lands on both. + /// OTOPCUA_EVENT_PERF_ROUNDS sets the rounds (default 2). + /// + [Theory] + [InlineData(50, 1000u, false)] + [InlineData(200, 1000u, false)] + [InlineData(200, 1000u, true)] + [InlineData(50, 10000u, false)] + public async Task Ring_vs_sdk_event_queue(int n, uint queueSize, bool acl) + { + RequireOptIn(); + var rounds = int.TryParse(Environment.GetEnvironmentVariable("OTOPCUA_EVENT_PERF_ROUNDS"), out var r) && r > 0 ? r : 2; + try + { + for (var round = 0; round < rounds; round++) + { + foreach (var ring in round % 2 == 0 ? new[] { false, true } : [true, false]) + { + OtOpcUaSdkServer.RingEventQueues = ring; + await RunAsync(n, acl, stalled: true, serverObjectProbe: false, queueSize, ring ? "ring" : "sdk "); + } + } + } + finally + { + OtOpcUaSdkServer.RingEventQueues = true; + } + } + private static async Task RunAsync( - int n, bool acl, bool stalled, bool serverObjectProbe, uint queueSize = DefaultQueueSize) + int n, bool acl, bool stalled, bool serverObjectProbe, uint queueSize = DefaultQueueSize, string? queueKind = null) { RequireOptIn(); var ct = TestContext.Current.CancellationToken; @@ -195,7 +230,7 @@ public sealed class FolderEventDeliveryPerfTests var deferredPeak = rig.Nm.DeferredEventReports; var row = new StringBuilder(); row.Append(CultureInfo.InvariantCulture, - $"{(serverObjectProbe ? "i=2253" : "folder"),-6} N={n,-3} acl={(acl ? "on " : "off")} {(stalled ? "stalled " : "publishing")} q={queueSize,-5} | "); + $"{(serverObjectProbe ? "i=2253" : "folder"),-6} N={n,-3} acl={(acl ? "on " : "off")} {(stalled ? "stalled " : "publishing")} q={queueSize,-5} {(queueKind is null ? "" : queueKind + " ")}| "); row.Append(CultureInfo.InvariantCulture, $"lock hold us p50 {Pct(holds, 50),8:F1} p90 {Pct(holds, 90),8:F1} p99 {Pct(holds, 99),8:F1} max {holds.Max(),9:F1} mean {holds.Average(),8:F1}"); row.Append(CultureInfo.InvariantCulture, $" | per-item mean {(n > 0 ? holds.Average() / n : 0),6:F2} us"); if (expectFolder) diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/RingEventQueueWiringTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/RingEventQueueWiringTests.cs new file mode 100644 index 000000000..2174f71e7 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/RingEventQueueWiringTests.cs @@ -0,0 +1,36 @@ +using Opc.Ua.Server; +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +/// +/// Gitea #870: the running server's monitored-item queue factory is (every event +/// item's queue is a ; the SDK's always asks +/// the server's factory), and data-change queues are still the SDK's own. +/// +// Shares a collection with the other user of OtOpcUaSdkServer.RingEventQueues (a process-wide switch the perf +// harness flips), so the wiring assertion never observes it switched off (#870 review). +[Collection(RingEventQueueSwitchCollection.Name)] +public sealed class RingEventQueueWiringTests +{ + [Fact] + public async Task The_server_creates_ring_event_queues_and_sdk_data_change_queues() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("RingQueue870", encrypted: false, ct); + + var factory = rig.Server.CurrentInstance.MonitoredItemQueueFactory.ShouldBeOfType(); + factory.SupportsDurableQueues.ShouldBeFalse(); + factory.CreateEventQueue(false, 1).ShouldBeOfType(); + factory.CreateDataChangeQueue(false, 2).ShouldBeOfType(); + } +} + +/// Serialises the tests that read or flip OtOpcUaSdkServer.RingEventQueues. +[CollectionDefinition(Name)] +public sealed class RingEventQueueSwitchCollection +{ + /// The collection name. + public const string Name = "RingEventQueueSwitch"; +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/RingEventMonitoredItemQueueParityTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/RingEventMonitoredItemQueueParityTests.cs new file mode 100644 index 000000000..677b1f573 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/RingEventMonitoredItemQueueParityTests.cs @@ -0,0 +1,308 @@ +using Opc.Ua; +using Opc.Ua.Server; +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests; + +/// +/// Gitea #870: must be observably identical to the SDK's +/// . Every test drives the SDK queue and the ring with the same operations and +/// compares every result — directly at the queue, and through the SDK's own (which owns +/// DiscardOldest, the overflow-event positions and the discarded/overflow counters) built over each factory. +/// +public sealed class RingEventMonitoredItemQueueParityTests +{ + private static readonly ITelemetryContext Telemetry = DefaultTelemetry.Create(_ => { }); + + private sealed class Target : IFilterTarget + { + public bool IsTypeOf(IFilterContext context, NodeId typeDefinitionId) => false; + + public Variant GetAttributeValue(IFilterContext context, NodeId typeDefinitionId, + ArrayOf relativePath, uint attributeId, NumericRange indexRange) => Variant.Null; + } + + private static readonly Target[] Targets = Enumerable.Range(0, 6).Select(_ => new Target()).ToArray(); + + private static EventFieldList NewEvent(Random random) => new() { Handle = Targets[random.Next(Targets.Length)] }; + + /// Runs on both, asserting the same outcome (value or exception type + status). + private static T Both(Func action, IEventMonitoredItemQueue sdk, + IEventMonitoredItemQueue ring, string step) + { + Exception? sdkError = null, ringError = null; + T sdkResult = default!, ringResult = default!; + try { sdkResult = action(sdk); } catch (Exception e) { sdkError = e; } + try { ringResult = action(ring); } catch (Exception e) { ringError = e; } + + (ringError?.GetType()).ShouldBe(sdkError?.GetType(), step); + if (sdkError is ServiceResultException s) + ((ServiceResultException)ringError!).StatusCode.ShouldBe(s.StatusCode, step); + if (sdkError is null) ringResult.ShouldBe(sdkResult, step); + ring.ItemsInQueue.ShouldBe(sdk.ItemsInQueue, step); + ring.QueueSize.ShouldBe(sdk.QueueSize, step); + return sdkResult; + } + + [Fact] + public void Initial_state_matches() + { + using var sdk = new EventMonitoredItemQueue(false, 7, Telemetry); + using var ring = new RingEventMonitoredItemQueue(7); + ring.MonitoredItemId.ShouldBe(sdk.MonitoredItemId); + ring.IsDurable.ShouldBe(sdk.IsDurable); + ring.QueueSize.ShouldBe(sdk.QueueSize); + ring.ItemsInQueue.ShouldBe(sdk.ItemsInQueue); + ring.Dequeue(out var value).ShouldBeFalse(); + value.ShouldBeNull(); + } + + [Fact] + public void Enqueue_at_queue_size_zero_throws_the_same_BadInternalError() + { + var sdk = new EventMonitoredItemQueue(false, 1, Telemetry); + var ring = new RingEventMonitoredItemQueue(1); + Both(q => { q.Enqueue(new EventFieldList()); return 0; }, sdk, ring, "enqueue at 0"); + Should.Throw(() => ring.Enqueue(new EventFieldList())).StatusCode + .ShouldBe(StatusCodes.BadInternalError); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public void Shrink_keeps_newest_or_oldest_and_dequeue_order_is_fifo(bool discardOldest) + { + var sdk = new EventMonitoredItemQueue(false, 1, Telemetry); + var ring = new RingEventMonitoredItemQueue(1); + var events = Enumerable.Range(0, 10).Select(_ => new EventFieldList()).ToArray(); + sdk.SetQueueSize(10, true); + ring.SetQueueSize(10, true); + // Wrap the ring first so the shrink crosses the buffer end. + for (var i = 0; i < 6; i++) { sdk.Enqueue(new EventFieldList()); ring.Enqueue(new EventFieldList()); } + for (var i = 0; i < 6; i++) { sdk.Dequeue(out _); ring.Dequeue(out _); } + foreach (var e in events) { sdk.Enqueue(e); ring.Enqueue(e); } + + sdk.SetQueueSize(4, discardOldest); + ring.SetQueueSize(4, discardOldest); + + var expected = discardOldest ? events[6..] : events[..4]; + foreach (var e in expected) + { + sdk.Dequeue(out var a).ShouldBeTrue(); + ring.Dequeue(out var b).ShouldBeTrue(); + a.ShouldBeSameAs(e); + b.ShouldBeSameAs(e); + } + + ring.Dequeue(out _).ShouldBeFalse(); + sdk.Dequeue(out _).ShouldBeFalse(); + } + + [Fact] + public void Enqueue_at_a_full_queue_drops_the_oldest() + { + var ring = new RingEventMonitoredItemQueue(1); + var sdk = new EventMonitoredItemQueue(false, 1, Telemetry); + ring.SetQueueSize(3, false); + sdk.SetQueueSize(3, false); + var events = Enumerable.Range(0, 5).Select(_ => new EventFieldList()).ToArray(); + foreach (var e in events) { ring.Enqueue(e); sdk.Enqueue(e); } + foreach (var e in events[2..]) + { + ring.Dequeue(out var b).ShouldBeTrue(); + sdk.Dequeue(out var a).ShouldBeTrue(); + b.ShouldBeSameAs(e); + a.ShouldBeSameAs(e); + } + } + + [Fact] + public void Duplicate_check_covers_exactly_the_newest_1000_entries() + { + var sdk = new EventMonitoredItemQueue(false, 1, Telemetry); + var ring = new RingEventMonitoredItemQueue(1); + sdk.SetQueueSize(3000, true); + ring.SetQueueSize(3000, true); + var old = new Target(); + var edge = new Target(); + void Add(EventFieldList e) { sdk.Enqueue(e); ring.Enqueue(e); } + Add(new EventFieldList { Handle = old }); + Add(new EventFieldList { Handle = edge }); + for (var i = 0; i < 999; i++) Add(new EventFieldList { Handle = Targets[0] }); + Add(null!); // a null entry is skipped, as the SDK skips it + + // 1002 entries: `old` is the 1002nd newest, `edge` the 1001st — both outside the 1000-entry window. + Both(q => q.IsEventContainedInQueue(old), sdk, ring, "old").ShouldBeFalse(); + Both(q => q.IsEventContainedInQueue(edge), sdk, ring, "edge").ShouldBeFalse(); + sdk.Dequeue(out _); ring.Dequeue(out _); + sdk.Dequeue(out _); ring.Dequeue(out _); + Add(new EventFieldList { Handle = edge }); + Both(q => q.IsEventContainedInQueue(edge), sdk, ring, "newest").ShouldBeTrue(); + Both(q => q.IsEventContainedInQueue(Targets[0]), sdk, ring, "middle").ShouldBeTrue(); + Both(q => q.IsEventContainedInQueue(Targets[1]), sdk, ring, "absent").ShouldBeFalse(); + } + + /// Upper boundary of the window: the entry exactly 1000th from newest is still found; 1001st is not. + [Fact] + public void Duplicate_check_includes_the_1000th_newest_entry() + { + var sdk = new EventMonitoredItemQueue(false, 1, Telemetry); + var ring = new RingEventMonitoredItemQueue(1); + sdk.SetQueueSize(3000, true); + ring.SetQueueSize(3000, true); + var outside = new Target(); + var boundary = new Target(); + void Add(EventFieldList e) { sdk.Enqueue(e); ring.Enqueue(e); } + Add(new EventFieldList { Handle = outside }); + Add(new EventFieldList { Handle = boundary }); + for (var i = 0; i < 999; i++) Add(new EventFieldList { Handle = Targets[0] }); + + // 1001 entries: `boundary` is the 1000th newest (inside), `outside` the 1001st. + Both(q => q.IsEventContainedInQueue(boundary), sdk, ring, "boundary").ShouldBeTrue(); + Both(q => q.IsEventContainedInQueue(outside), sdk, ring, "outside").ShouldBeFalse(); + } + + /// Random interleavings of every queue operation, at small sizes (wrap-around, grow, shrink, size 0). + [Theory] + [MemberData(nameof(Seeds))] + public void Random_operation_sequences_match_the_sdk_queue(int seed) + { + var random = new Random(seed); + var sdk = new EventMonitoredItemQueue(false, 3, Telemetry); + var ring = new RingEventMonitoredItemQueue(3); + for (var step = 0; step < 2000; step++) + { + var label = $"seed {seed} step {step}"; + switch (random.Next(10)) + { + case < 5: + var e = random.Next(50) == 0 ? null! : NewEvent(random); + Both(q => { q.Enqueue(e); return 0; }, sdk, ring, label); + break; + case < 8: + var (sdkHad, sdkValue) = (sdk.Dequeue(out var a), a); + var (ringHad, ringValue) = (ring.Dequeue(out var b), b); + ringHad.ShouldBe(sdkHad, label); + ringValue.ShouldBeSameAs(sdkValue, label); + break; + case 8: + var size = (uint)(random.Next(8) == 0 ? 0 : random.Next(1, 40)); + var discardOldest = random.Next(2) == 0; + Both(q => { q.SetQueueSize(size, discardOldest); return 0; }, sdk, ring, label); + break; + default: + var t = Targets[random.Next(Targets.Length)]; + Both(q => q.IsEventContainedInQueue(t), sdk, ring, label); + break; + } + } + + // Drain: the same remaining contents in the same order. + while (sdk.Dequeue(out var a)) + { + ring.Dequeue(out var b).ShouldBeTrue(); + b.ShouldBeSameAs(a); + } + + ring.Dequeue(out _).ShouldBeFalse(); + } + + public static TheoryData Seeds() => new(Enumerable.Range(1, 40)); + + /// + /// The SDK's over the SDK factory vs over : + /// QueueEvent (incl. the DiscardOldest=false refusal), SetQueueOverflowIfFull, SetQueueSize with DiscardOldest + /// flips, Publish with small batches (dequeue mid-stream) — the published events, the overflow-event-due flags, + /// the Overflow property and the discarded/overflow callbacks must all agree. + /// + [Theory] + [MemberData(nameof(Seeds))] + public void The_sdk_event_queue_handler_behaves_identically_over_either_queue(int seed) + { + var random = new Random(seed); + var counts = new int[4]; // sdk discarded, sdk overflow, ring discarded, ring overflow + using var sdkFactory = new MonitoredItemQueueFactory(Telemetry); + using var ringFactory = new RingEventQueueFactory(new MonitoredItemQueueFactory(Telemetry)); + using var sdk = new EventQueueHandler(false, sdkFactory, 9, Telemetry, () => counts[0]++, () => counts[1]++); + using var ring = new EventQueueHandler(false, ringFactory, 9, Telemetry, () => counts[2]++, () => counts[3]++); + var context = new OperationContext(new RequestHeader(), null, RequestType.Publish, RequestLifetime.None); + + var initialOldest = random.Next(2) == 0; + sdk.SetQueueSize(10, initialOldest); + ring.SetQueueSize(10, initialOldest); + + for (var step = 0; step < 1500; step++) + { + var label = $"seed {seed} step {step}"; + switch (random.Next(12)) + { + case < 6: + var e = NewEvent(random); + Exception? a = null, b = null; + try { sdk.QueueEvent(e); } catch (Exception x) { a = x; } + try { ring.QueueEvent(e); } catch (Exception x) { b = x; } + (b?.GetType()).ShouldBe(a?.GetType(), label); + break; + case 6: + ring.SetQueueOverflowIfFull().ShouldBe(sdk.SetQueueOverflowIfFull(), label); + break; + case 7: + var size = (uint)random.Next(1, 25); + var discardOldest = random.Next(2) == 0; + sdk.SetQueueSize(size, discardOldest); + ring.SetQueueSize(size, discardOldest); + break; + case < 10: + var max = (uint)random.Next(1, 6); + var sdkOut = new Queue(); + var ringOut = new Queue(); + var sdkN = sdk.Publish(context, sdkOut, max); + var ringN = ring.Publish(context, ringOut, max); + ringN.ShouldBe(sdkN, label); + ringOut.Count.ShouldBe(sdkOut.Count, label); + while (sdkOut.Count > 0) ringOut.Dequeue().ShouldBeSameAs(sdkOut.Dequeue(), label); + break; + default: + var t = Targets[random.Next(Targets.Length)]; + ring.IsEventContainedInQueue(t).ShouldBe(sdk.IsEventContainedInQueue(t), label); + break; + } + + ring.ItemsInQueue.ShouldBe(sdk.ItemsInQueue, label); + ring.Overflow.ShouldBe(sdk.Overflow, label); + counts[2].ShouldBe(counts[0], label); + counts[3].ShouldBe(counts[1], label); + } + + // Positive control: the sequence really overflowed (discards and overflow events happened). + counts[0].ShouldBeGreaterThan(0); + counts[1].ShouldBeGreaterThan(0); + } + + [Fact] + public void Factory_supplies_ring_event_queues_and_delegates_everything_else() + { + using var factory = new RingEventQueueFactory(new MonitoredItemQueueFactory(Telemetry)); + factory.SupportsDurableQueues.ShouldBeFalse(); + factory.CreateEventQueue(false, 4).ShouldBeOfType().MonitoredItemId.ShouldBe(4u); + factory.CreateDataChangeQueue(false, 5).ShouldBeOfType(); + // A durable request reaches the SDK factory, which refuses it as before. + Should.Throw(() => factory.CreateEventQueue(true, 6)); + Should.Throw(() => new MonitoredItemQueueFactory(Telemetry).CreateEventQueue(true, 6)); + } + + [Fact] + public void Buffer_grows_lazily_and_is_released_on_shrink() + { + var ring = new RingEventMonitoredItemQueue(1); + ring.SetQueueSize(10000, true); + ring.Capacity.ShouldBe(0); + for (var i = 0; i < 20000; i++) ring.Enqueue(new EventFieldList()); + ring.Capacity.ShouldBe(10000); + ring.ItemsInQueue.ShouldBe(10000); + ring.SetQueueSize(10, true); + ring.ItemsInQueue.ShouldBe(10); + ring.Capacity.ShouldBeLessThanOrEqualTo(20); + } +}