From 71b3b7f35f9cd1ae359028733f45420d2224cb85 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 15:06:26 -0500 Subject: [PATCH 1/8] fix(galaxy): #866 map MxStatus failure details by the MxStatusDetail vocabulary, not as OPC DA quality The gateway copies MXSTATUS_PROXY.detail verbatim, so a failure row's detail is always an MxStatusDetail code. FromMxStatus routed any detail <= 0xFF through the OPC DA quality table: MX_E_PlatformCommunicationError (2) on a stopped platform's ScanState logged "Unrecognised MXAccess quality byte 0x02" and published generic Bad (VD03 live capture), MX_E_InvalidObjectId (8) became BadNotConnected, MX_E_Overflow (20) BadCommunicationError, and a detail in the Uncertain band published Uncertain for a failure. Every non-zero detail now maps per the gateway.md vocabulary (timeout, communication, configuration, type, range, access, initialising), then by category; the category-fallback warning logs once per (detail, category). Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../Runtime/StatusCodeMap.cs | 75 +++++++++++-------- .../Runtime/StatusCodeMapTests.cs | 65 +++++++++++++--- 2 files changed, 100 insertions(+), 40 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/StatusCodeMap.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/StatusCodeMap.cs index 99dde2880..7303a8591 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/StatusCodeMap.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/StatusCodeMap.cs @@ -110,8 +110,8 @@ internal static class StatusCodeMap /// (category == OK AND success != 0) /// as the canonical success test — the proto contract explicitly documents that /// success is NOT a boolean and must not be checked in isolation; category is - /// the authoritative indicator. On failure, the detail byte (OPC DA quality substatus) - /// drives the specific code, with a transport-error fallback for pre-MXAccess failures. + /// the authoritative indicator. On failure, the MxStatusDetail code drives the specific + /// code, then the category, with a transport-error fallback for pre-MXAccess failures. /// /// The MX status proxy to convert, or null for Good status. /// Optional logger for diagnostics on unknown status codes. @@ -121,21 +121,14 @@ internal static class StatusCodeMap if (status is null) return Good; if (status.IsSuccess()) return Good; - // A detail in [1, 255] is an OPC DA quality substatus (read/data-change rows). Anything - // larger is an MXAccess MxStatus detail code — the write-completion rows the gateway - // correlates onto write replies carry these (e.g. 1007 operational, 1008 security) — and - // truncating one into the quality table maps a REFUSED WRITE to Good whenever the low byte - // lands in the Good band (1007 → 239 ≥ 192). Route those on the status category instead. - var detail = status.Detail; - if (detail is > 0 and <= 0xFF) - { - var mapped = FromQualityByte((byte)detail, logger); - // Invariant: a row that is not IsSuccess() must never surface as Good — a - // contradictory Good-band substatus on a failure row falls back to the category. - if ((mapped >> 30) != 0u) return mapped; - return FromCategory(status, logger); - } - if (detail > 0xFF) return FromCategory(status, logger); + // The detail is MXAccess's MxStatusDetail code verbatim (mxaccessgw's MxStatusProxyConverter copies + // MXSTATUS_PROXY.detail straight through; vocabulary in its gateway.md) — NEVER an OPC DA quality + // byte. It used to be routed through the quality table whenever it was <= 0xFF, which mis-named + // every low code: MX_E_PlatformCommunicationError (2) logged as "quality byte 0x02" and fell to + // generic Bad (#866, a stopped platform's ScanState on VD03), MX_E_InvalidObjectId (8) surfaced as + // BadNotConnected, MX_E_Overflow (20) as BadCommunicationError. A failure row is also never Good or + // Uncertain, whatever its detail. + if (status.Detail != 0) return FromCategory(status, logger); // detected_by != Mxaccess (raw_detected_by != the MXAccess source enum) implies // the failure happened pre-call (gateway, worker, transport) — surface as a @@ -146,33 +139,47 @@ internal static class StatusCodeMap } /// - /// Map a failure row whose detail field is an MxStatus detail code rather than an OPC DA - /// quality byte (or contradicts the failure). Per-code first — the MxStatusDetail - /// vocabulary is published in mxaccessgw's gateway.md ("MxStatus Detail Vocabulary", - /// 75c71ad, lifted from Interop.aaMxDataConsumer.MxStatusDetail) — then the - /// category bucket. Codes ≤ 0xFF (e.g. 1 = MX_E_RequestTimedOut) never reach here: they are - /// value-ambiguous with the quality-byte space and take the quality path, which still lands - /// them in the Bad family. + /// Map a failure row by its MxStatusDetail code — the vocabulary is published in mxaccessgw's + /// gateway.md ("MxStatus Detail Vocabulary", lifted from Interop.aaMxDataConsumer.MxStatusDetail) + /// — then by the status category for codes with no specific OPC UA counterpart. /// private static uint FromCategory(MxStatusProxy status, ILogger? logger) { switch (status.Detail) { - case 1004: return BadOutOfRange; // MX_E_DataOutOfRange + case 1: return BadTimeout; // MX_E_RequestTimedOut + case 2: // MX_E_PlatformCommunicationError (live: stopped platform, #866) + case 5: return BadCommunicationError; // MX_E_EngineCommunicationError + case 3: // MX_E_InvalidPlatformId + case 4: // MX_E_InvalidEngineId + case 6: // MX_E_InvalidReference + case 8: // MX_E_InvalidObjectId + case 9: // MX_E_ObjectSignatureMismatch + case 10: // MX_E_AttributeSignatureMismatch + case 1000: // MX_E_InvalidPrimitiveId + case 1001: // MX_E_InvalidAttributeId + case 1002: return BadConfigurationError; // MX_E_InvalidPropertyId + case 13: // MX_E_WrongDataType + case 18: // MX_E_ConversionNotSupported + case 19: // MX_E_UnableToConvertString case 1005: return BadTypeMismatch; // MX_E_IncorrectDataType + case 20: // MX_E_Overflow + case 1004: return BadOutOfRange; // MX_E_DataOutOfRange case 1006: return BadNotReadable; // MX_E_NotReadable case 1007: return BadNotWritable; // MX_E_NotWriteable (live-captured: read-only attr) case 1008: // MX_E_WriteAccessDenied (live-captured: unadvised no-login write) case 1012: // MX_E_SecuredWrite (attribute requires a secured write) case 1013: // MX_E_VerifiedWrite (attribute requires a verified write) + case 1014: // MX_E_NoAlarmAckPrivilege case 1016: // MX_E_UserNotHavingAccessRights case 1017: return BadUserAccessDenied; // MX_E_VerifierNotHavingVerifyRights case 1010: // MX_E_ObjectInitializing (transient) case 1011: return BadOutOfService; // MX_E_EngineInitializing (transient) } - // Everything else — including 8000 MX_E_AutomationObjectSpecificError, deliberately, since - // the object supplies its own semantics — falls back to the category bucket. + // Everything else — resolving (11/12), repository/engine busy (7/25/26), version/command mismatches, + // index errors, 1009 unknown and, deliberately, 8000 MX_E_AutomationObjectSpecificError since the + // object supplies its own semantics — falls back to the category bucket. var code = status.Category switch { MxStatusCategory.SecurityError => BadUserAccessDenied, @@ -180,13 +187,19 @@ internal static class StatusCodeMap MxStatusCategory.ConfigurationError => BadConfigurationError, _ => Bad, }; - logger?.LogWarning( - "MxStatus failure row category {Category} detail {Detail} mapped to 0x{Code:X8} at category " + - "level. Field capture welcome — extend StatusCodeMap.FromCategory with the detail code.", - status.Category, status.Detail, code); + // Once per (detail, category) per process: these rows arrive on the data-change path, so a stopped + // engine's tags would otherwise log on every update. + if (logger is not null && UnmappedDetailsLogged.TryAdd((status.Detail, status.Category), 0)) + logger.LogWarning( + "MxStatus failure row category {Category} detail {Detail} ({DiagnosticText}) mapped to 0x{Code:X8} at " + + "category level. Field capture welcome — extend StatusCodeMap.FromCategory with the detail code.", + status.Category, status.Detail, status.DiagnosticText, code); return code; } + private static readonly System.Collections.Concurrent.ConcurrentDictionary<(int, MxStatusCategory), byte> + UnmappedDetailsLogged = new(); + /// /// Convert an OPC UA status-code uint back to the OPC DA quality category /// byte — Good=192, Uncertain=64, Bad=0 — by extracting the top-two bits of the diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/StatusCodeMapTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/StatusCodeMapTests.cs index 4b4d9084a..83c24ae09 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/StatusCodeMapTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/StatusCodeMapTests.cs @@ -84,12 +84,60 @@ public sealed class StatusCodeMapTests StatusCodeMap.FromMxStatus(s).ShouldNotBe(StatusCodeMap.Good); } - /// Verifies that known detail codes map to their specific status codes. + /// + /// #866: a low detail is an MxStatusDetail code, not an OPC DA quality byte. Detail 8 is + /// MX_E_InvalidObjectId — it used to surface as BadNotConnected (quality byte 8). + /// [Fact] public void FromMxStatus_SuccessZeroAndCategoryNotOk_DetailKnown_MapsToSpecificCode() { - var s = new MxStatusProxy { Success = 0, Category = MxStatusCategory.OperationalError, Detail = 8 /* Bad_NotConnected */ }; - StatusCodeMap.FromMxStatus(s).ShouldBe(StatusCodeMap.BadNotConnected); + var s = new MxStatusProxy { Success = 0, Category = MxStatusCategory.OperationalError, Detail = 8 }; + StatusCodeMap.FromMxStatus(s).ShouldBe(StatusCodeMap.BadConfigurationError); + } + + /// + /// #866 live capture (VD03 2026-11-11): reading JoePlatform.ScanState on a deployed but + /// not-running platform returns a failure row with detail 2 = MX_E_PlatformCommunicationError, + /// which logged "Unrecognised MXAccess quality byte 0x02" and published generic Bad. + /// + [Fact] + public void FromMxStatus_PlatformCommunicationError_IsCommunicationError_AndDoesNotLogAsQuality() + { + var logger = new CapturingLogger(); + var s = new MxStatusProxy { Success = 0, Category = MxStatusCategory.CommunicationError, Detail = 2 }; + StatusCodeMap.FromMxStatus(s, logger).ShouldBe(StatusCodeMap.BadCommunicationError); + logger.Entries.ShouldBeEmpty(); + } + + /// Low MxStatusDetail codes map per the gateway.md vocabulary regardless of category. + [Theory] + [InlineData(1, 0x800A0000u)] // MX_E_RequestTimedOut → BadTimeout + [InlineData(2, 0x80050000u)] // MX_E_PlatformCommunicationError → BadCommunicationError + [InlineData(5, 0x80050000u)] // MX_E_EngineCommunicationError → BadCommunicationError + [InlineData(3, 0x80890000u)] // MX_E_InvalidPlatformId → BadConfigurationError + [InlineData(4, 0x80890000u)] // MX_E_InvalidEngineId (was BadConfigurationError by quality-byte accident) + [InlineData(6, 0x80890000u)] // MX_E_InvalidReference + [InlineData(12, 0x80000000u)] // MX_E_ResolvingObject → category (was BadDeviceFailure) + [InlineData(13, 0x80740000u)] // MX_E_WrongDataType → BadTypeMismatch + [InlineData(16, 0x80000000u)] // MX_E_IndexOutOfOrder → category (was BadSensorFailure) + [InlineData(19, 0x80740000u)] // MX_E_UnableToConvertString → BadTypeMismatch + [InlineData(20, 0x803C0000u)] // MX_E_Overflow → BadOutOfRange (was BadCommunicationError) + [InlineData(24, 0x80000000u)] // MX_E_LmxInvalidCommand → category (was BadOutOfService) + [InlineData(32, 0x80000000u)] // not a vocabulary code → category (was BadWaitingForInitialData) + public void FromMxStatus_LowDetailCodes_MapByVocabulary(int detail, uint expected) + { + var s = new MxStatusProxy { Success = 0, Category = MxStatusCategory.OperationalError, Detail = detail }; + StatusCodeMap.FromMxStatus(s).ShouldBe(expected); + } + + /// The category fallback logs once per (detail, category), not on every data change. + [Fact] + public void FromMxStatus_CategoryFallback_LogsOncePerDetailAndCategory() + { + var logger = new CapturingLogger(); + var s = new MxStatusProxy { Success = 0, Category = MxStatusCategory.SoftwareError, Detail = 23 /* MX_E_LmxVersionMismatch */ }; + for (var i = 0; i < 5; i++) StatusCodeMap.FromMxStatus(s, logger).ShouldBe(StatusCodeMap.Bad); + logger.Entries.Count.ShouldBe(1); } /// Verifies that non-zero DetectedBy maps to communication error when detail is zero. @@ -155,9 +203,8 @@ public sealed class StatusCodeMapTests } /// - /// Invariant: a row that is not IsSuccess() must NEVER map to Good — a Good-band quality - /// byte (e.g. 216 = GoodLocalOverride) on a failure row is contradictory and falls back to - /// the category route. Uncertain-band substatuses remain allowed (uncertain is not Good). + /// Invariant: a row that is not IsSuccess() must NEVER map to Good — a detail that happens to sit in + /// the OPC DA Good band (216) is not a quality byte and routes on the category. /// [Fact] public void FromMxStatus_FailureRow_GoodBandQualityByte_NeverMapsGood() @@ -166,12 +213,12 @@ public sealed class StatusCodeMapTests StatusCodeMap.FromMxStatus(s).ShouldBe(StatusCodeMap.BadUserAccessDenied); } - /// An Uncertain-band quality substatus on a failure row still maps to its specific Uncertain code. + /// A failure row is Bad: a detail in the OPC DA Uncertain band (68) is not a quality byte (#866). [Fact] - public void FromMxStatus_FailureRow_UncertainBandQualityByte_KeepsUncertainCode() + public void FromMxStatus_FailureRow_UncertainBandDetail_IsBad() { var s = new MxStatusProxy { Success = 0, Category = MxStatusCategory.OperationalError, Detail = 68 }; - StatusCodeMap.FromMxStatus(s).ShouldBe(StatusCodeMap.UncertainLastUsableValue); + StatusCodeMap.FromMxStatus(s).ShouldBe(StatusCodeMap.Bad); } /// Verifies that top two bits of status codes follow OPC UA convention. From 9aaab9ee421504b2cfec84e4cb03df82b6272888 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 15:09:20 -0500 Subject: [PATCH 2/8] test: #871 fix two hosted-CI races that failed master runs 1218-1220 - MeshPskProviderRefreshTests: GatedResolver.Entered was set by the priming refresh and never reset, so Wait() returned before the held refresh reached the store and the mid-refresh call-count snapshot raced its increment (expected 1, was 2). Hold() now resets it. - FailedCredentialLockoutTests all-bad burst: the 15 waiters could queue before a slot holder ran the increment after BeginAsync (authenticated 0 vs 5). The wait now covers both conditions; the security assertions are unchanged. Both looped 20x locally, 0 failures. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../Communication/MeshPskProviderRefreshTests.cs | 8 +++++++- .../FailedCredentialLockoutTests.cs | 4 +++- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshPskProviderRefreshTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshPskProviderRefreshTests.cs index 02aafc1b6..36c209b7d 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshPskProviderRefreshTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/MeshPskProviderRefreshTests.cs @@ -232,7 +232,13 @@ public sealed class MeshPskProviderRefreshTests public int Calls => Volatile.Read(ref _calls); public ManualResetEventSlim Entered { get; } = new(); - public void Hold() => _gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + // Reset Entered too: the priming refresh already set it, so without the reset Wait() returned before the + // held call had even reached the store and the call-count snapshot raced its increment (hosted-CI flake). + public void Hold() + { + Entered.Reset(); + _gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + } public void Release() => _gate?.TrySetResult(); public async Task GetAsync(SecretName name, CancellationToken ct) diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/FailedCredentialLockoutTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/FailedCredentialLockoutTests.cs index 0c0d5cfdf..85a4741e4 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/FailedCredentialLockoutTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/FailedCredentialLockoutTests.cs @@ -162,7 +162,9 @@ public sealed class FailedCredentialLockoutTests lockout.Complete(Client, succeeded: false); }, Ct)).ToArray(); - await WaitUntilAsync(() => lockout.WaitingCount(Client) == 15); + // Wait for the slot holders' increments too: the 15 waiters can queue before a slot holder's thread has run + // the line after BeginAsync, which read authenticated == 0 on the 2-core hosted runner. + await WaitUntilAsync(() => lockout.WaitingCount(Client) == 15 && Volatile.Read(ref authenticated) == 5); authenticated.ShouldBe(5); release.SetResult(); await Task.WhenAll(attempts); From 875b82175d4029538f4244fb44eee77f9a684d57 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 15:14:11 -0500 Subject: [PATCH 3/8] fix(galaxy): #866 grade OK-status data changes by the item's OPC DA quality The gateway fills Statuses on every data change (live OnDataChange, value-cache replay and session snapshot all carry Quality too), and EventPump used the status row whenever one was present, so Quality was never read: an off-scan, Bad or Uncertain attribute with an OK status row was published Good. A failure status row still wins; otherwise the quality byte grades the value (legacy parity: quality >= 192 Good). Live, read-only on VD03 (DEV): healthy ScanState and _TestArea attributes stay Good (0x00000000) via quality 192; the stopped JoePlatform/JoeAppEngine ScanState now publish BadCommunicationError (0x80050000) instead of generic Bad, with no "unrecognised quality byte" warning. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../Runtime/EventPump.cs | 8 +- .../Runtime/EventPumpQualityTests.cs | 84 +++++++++++++++++++ 2 files changed, 91 insertions(+), 1 deletion(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpQualityTests.cs diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs index cdb36a264..20a9a4597 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs @@ -2,6 +2,7 @@ using System.Diagnostics.Metrics; using System.Threading.Channels; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.MxGateway.Client; using ZB.MOM.WW.MxGateway.Contracts.Proto; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; @@ -512,7 +513,12 @@ internal sealed class EventPump : IAsyncDisposable ServerTimestampUtc: DateTime.UtcNow); } - var statusCode = ev.Statuses.Count > 0 + // A failure status row wins (it names why the read failed); otherwise the item's OPC DA quality grades + // the value. The gateway fills Statuses on every data change, so keying on "any statuses present" meant + // Quality was never read and an off-scan / Bad / Uncertain attribute with an OK status row was + // published Good (#866 review). Every gateway path (live OnDataChange, value-cache replay, session + // snapshot) carries Quality. + var statusCode = ev.Statuses.Count > 0 && !ev.Statuses[0].IsSuccess() ? StatusCodeMap.FromMxStatus(ev.Statuses[0], _logger) : StatusCodeMap.FromQualityByte((byte)(ev.Quality & 0xFF), _logger); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpQualityTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpQualityTests.cs new file mode 100644 index 000000000..f11b46396 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpQualityTests.cs @@ -0,0 +1,84 @@ +using System.Runtime.CompilerServices; +using System.Threading.Channels; +using Google.Protobuf.WellKnownTypes; +using Shouldly; +using Xunit; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; + +/// +/// #866 review: the gateway fills Statuses on every data change, and the pump used the status row +/// whenever one was present, so the item's OPC DA quality was never read. An off-scan / Bad / Uncertain +/// attribute carrying an OK status row was published Good. A failure status row still wins. +/// +public sealed class EventPumpQualityTests +{ + private const int WaitMs = 10_000; + + private static MxStatusProxy Ok => new() { Success = 1, Category = MxStatusCategory.Ok }; + + private sealed class ChannelSubscriber : IGalaxySubscriber + { + private readonly Channel _events = Channel.CreateUnbounded(); + + public Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + => Task.FromResult>([]); + + public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) => Task.CompletedTask; + + public async IAsyncEnumerable StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken) + { + await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + yield return ev; + } + + public ValueTask EmitAsync(int handle, int quality, MxStatusProxy status) + { + var ev = new MxEvent + { + Family = MxEventFamily.OnDataChange, + ItemHandle = handle, + Value = new MxValue { DoubleValue = 1 }, + Quality = quality, + SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow), + }; + ev.Statuses.Add(status); + return _events.Writer.WriteAsync(ev); + } + } + + private static async Task PublishedStatusAsync(int quality, MxStatusProxy status) + { + var subscriber = new ChannelSubscriber(); + var registry = new SubscriptionRegistry(); + registry.Register(1, [new TagBinding("Tank01.Level", 7)]); + var delivered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var pump = new EventPump(subscriber, registry, channelCapacity: 4, clientName: "Quality"); + pump.OnDataChange += (_, e) => delivered.TrySetResult(e); + pump.Start(); + + await subscriber.EmitAsync(7, quality, status); + (await Task.WhenAny(delivered.Task, Task.Delay(WaitMs, TestContext.Current.CancellationToken))).ShouldBe(delivered.Task); + return (await delivered.Task).Snapshot.StatusCode; + } + + [Theory] + [InlineData(192, 0x00000000u)] // Good + [InlineData(24, 0x808D0000u)] // Bad OutOfService (off scan) + [InlineData(8, 0x808A0000u)] // Bad NotConnected + [InlineData(68, 0x40900000u)] // Uncertain LastUsableValue + public async Task An_ok_status_row_is_graded_by_the_item_quality(int quality, uint expected) + => (await PublishedStatusAsync(quality, Ok)).ShouldBe(expected); + + [Fact] + public async Task A_failure_status_row_wins_over_a_good_quality() + { + var failure = new MxStatusProxy { Success = 0, Category = MxStatusCategory.CommunicationError, Detail = 2 }; + (await PublishedStatusAsync(192, failure)).ShouldBe(StatusCodeMap.BadCommunicationError); + } +} From 970f2159ad77bcc9b47d55c42741b90100b4c27c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 15:14:22 -0500 Subject: [PATCH 4/8] docs(galaxy): IsSuccess checks the status category only Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/StatusCodeMap.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/StatusCodeMap.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/StatusCodeMap.cs index 7303a8591..c316a2bcd 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/StatusCodeMap.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/StatusCodeMap.cs @@ -107,8 +107,8 @@ internal static class StatusCodeMap /// /// Map a gateway-reported to OPC UA StatusCode. Uses - /// (category == OK AND success != 0) - /// as the canonical success test — the proto contract explicitly documents that + /// (category == OK; the client library checks the + /// category only) as the canonical success test — the proto contract explicitly documents that /// success is NOT a boolean and must not be checked in isolation; category is /// the authoritative indicator. On failure, the MxStatusDetail code drives the specific /// code, then the category, with a transport-error fallback for pre-MXAccess failures. From 1a08b4f746d2a4f7c7d7106fedc70363885c6ab1 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 15:15:05 -0500 Subject: [PATCH 5/8] fix(galaxy): #872 detail-0 failure rows route on category, not detected_by Every raw detected_by value names an MXAccess layer (requesting/responding LMX, NMX, automation object); the gateway's own failures arrive as ProtocolStatus. The 'non-zero source = transport failure' rule made a failure the responding object detected a comms error. Item 1 of #872 (Warning/Pending categories) stays open pending a captured row. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../Runtime/StatusCodeMap.cs | 14 ++++++-------- .../Runtime/StatusCodeMapTests.cs | 16 +++++++++++----- 2 files changed, 17 insertions(+), 13 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/StatusCodeMap.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/StatusCodeMap.cs index c316a2bcd..0fe9b7b3a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/StatusCodeMap.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/StatusCodeMap.cs @@ -128,14 +128,12 @@ internal static class StatusCodeMap // generic Bad (#866, a stopped platform's ScanState on VD03), MX_E_InvalidObjectId (8) surfaced as // BadNotConnected, MX_E_Overflow (20) as BadCommunicationError. A failure row is also never Good or // Uncertain, whatever its detail. - if (status.Detail != 0) return FromCategory(status, logger); - - // detected_by != Mxaccess (raw_detected_by != the MXAccess source enum) implies - // the failure happened pre-call (gateway, worker, transport) — surface as a - // communication error rather than a generic Bad. - if (status.RawDetectedBy != 0) return BadCommunicationError; - - return Bad; + // + // Detail 0 routes the same way. It used to become BadCommunicationError whenever raw_detected_by != 0, on + // the belief that a non-zero source meant a pre-MXAccess transport failure — but every detected_by value + // names an MXAccess layer (0 requesting LMX, 1 responding LMX, 2/3 NMX, 4/5 automation object; the + // gateway's own failures arrive as ProtocolStatus, never as a status row), so the category decides (#872). + return FromCategory(status, logger); } /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/StatusCodeMapTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/StatusCodeMapTests.cs index 83c24ae09..64f99ced1 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/StatusCodeMapTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/StatusCodeMapTests.cs @@ -140,12 +140,18 @@ public sealed class StatusCodeMapTests logger.Entries.Count.ShouldBe(1); } - /// Verifies that non-zero DetectedBy maps to communication error when detail is zero. - [Fact] - public void FromMxStatus_SuccessZero_DetailZero_DetectedByNonZero_IsCommunicationError() + /// + /// #872: detected_by names an MXAccess layer (3 = responding NMX), not a transport failure, so a detail-0 + /// failure row routes on its category — a communication-category row is still BadCommunicationError. + /// + [Theory] + [InlineData(MxStatusCategory.Unspecified, 0x80000000u)] + [InlineData(MxStatusCategory.CommunicationError, 0x80050000u)] + [InlineData(MxStatusCategory.ConfigurationError, 0x80890000u)] + public void FromMxStatus_SuccessZero_DetailZero_RoutesByCategory_NotDetectedBy(MxStatusCategory category, uint expected) { - var s = new MxStatusProxy { Success = 0, Detail = 0, RawDetectedBy = 3 }; - StatusCodeMap.FromMxStatus(s).ShouldBe(StatusCodeMap.BadCommunicationError); + var s = new MxStatusProxy { Success = 0, Detail = 0, RawDetectedBy = 3, Category = category }; + StatusCodeMap.FromMxStatus(s).ShouldBe(expected); } /// Verifies that all-zero status maps to Bad status code. From bcde93056a80fd9271bea7a7c03b7216e7249260 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 15:15:45 -0500 Subject: [PATCH 6/8] perf(opcua): #870 ring-buffer event monitored-item queue Since #855 condition events are queued to folder event items under the node manager Lock. At a full DiscardOldest queue the SDK's EventQueueHandler dequeues the head of EventMonitoredItemQueue, a List.RemoveAt(0) shift of the whole queue (~1 us/item at QueueSize 1000, ~10 us at 10000). OtOpcUaSdkServer.CreateMonitoredItemQueueFactory now wraps the SDK factory in RingEventQueueFactory: event queues are RingEventMonitoredItemQueue (O(1) discard), data-change queues, durable requests and disposal go to the SDK factory unchanged. A factory set through StandardServer.MonitoredItemQueueFactory is used as-is. The SDK queue is storage only (decompiled 2.0.1): DiscardOldest, overflow-event positions and discarded/overflow counters live in EventQueueHandler, which is unchanged. The ring reproduces QueueSize 0 initial + BadInternalError enqueue, drop-oldest at Count == QueueSize, FIFO dequeue, SetQueueSize keep-newest / keep-oldest, the newest-1000 duplicate window via IsEventRaisedFor, no-op Dispose. RingEventMonitoredItemQueueParityTests drive both queues, and the SDK EventQueueHandler over both factories, with the same random sequences. FolderEventDeliveryPerfTests gains Ring_vs_sdk_event_queue (A/B in one process via the internal OtOpcUaSdkServer.RingEventQueues switch). Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .claude/rules/data-plane-auth.md | 2 +- .../OtOpcUaSdkServer.cs | 21 ++ .../RingEventMonitoredItemQueue.cs | 182 +++++++++++ .../FolderEventDeliveryPerfTests.cs | 36 ++- .../RingEventQueueWiringTests.cs | 25 ++ .../RingEventMonitoredItemQueueParityTests.cs | 288 ++++++++++++++++++ 6 files changed, 551 insertions(+), 3 deletions(-) create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/RingEventMonitoredItemQueue.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/RingEventQueueWiringTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/RingEventMonitoredItemQueueParityTests.cs 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..b101f5de6 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventDeliveryPerfTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventDeliveryPerfTests.cs @@ -100,8 +100,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 +227,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..f654cc186 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/RingEventQueueWiringTests.cs @@ -0,0 +1,25 @@ +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. +/// +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(); + } +} 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..b2d264ccf --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/RingEventMonitoredItemQueueParityTests.cs @@ -0,0 +1,288 @@ +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(); + } + + /// 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); + } +} From 36583ee6f4c9b03cbe8b9523aa6ed7937c7e3174 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 15:20:31 -0500 Subject: [PATCH 7/8] =?UTF-8?q?test(opcua):=20#870=20review=20=E2=80=94=20?= =?UTF-8?q?pin=20the=201000th-newest=20duplicate-window=20boundary;=20seri?= =?UTF-8?q?alise=20the=20ring-queue=20switch=20users?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../FolderEventDeliveryPerfTests.cs | 3 +++ .../RingEventQueueWiringTests.cs | 11 ++++++++++ .../RingEventMonitoredItemQueueParityTests.cs | 20 +++++++++++++++++++ 3 files changed, 34 insertions(+) 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 b101f5de6..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"; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/RingEventQueueWiringTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/RingEventQueueWiringTests.cs index f654cc186..2174f71e7 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/RingEventQueueWiringTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/RingEventQueueWiringTests.cs @@ -9,6 +9,9 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; /// 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] @@ -23,3 +26,11 @@ public sealed class RingEventQueueWiringTests 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 index b2d264ccf..677b1f573 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/RingEventMonitoredItemQueueParityTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/RingEventMonitoredItemQueueParityTests.cs @@ -143,6 +143,26 @@ public sealed class RingEventMonitoredItemQueueParityTests 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))] From 11ea972cd34518e63120a19cd37c29496b53712b Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 15:38:08 -0500 Subject: [PATCH 8/8] test(adminui): #871 reaper not-a-guid database name unique per run The fixed name lives on the SQL Server shared by every checkout and CI, so a concurrent run's create or finally-drop could collide (failed once in the local pre-push suite, 3/3 green on rerun; cause plausible, not proven). Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../SqlTestDatabaseReaperTests.cs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/SqlTestDatabaseReaperTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/SqlTestDatabaseReaperTests.cs index 92d676755..64b453b04 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/SqlTestDatabaseReaperTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/SqlTestDatabaseReaperTests.cs @@ -103,7 +103,9 @@ public sealed class SqlTestDatabaseReaperTests // what bounds the character set of every candidate: without it a database named // "_" + arbitrary text would be fed to the drop statement, and before the // statement was parameterised that was a second-order injection running as sa. - var notAGuid = $"{AdminUiSqlServer.TestDatabasePrefix}20250101000000_notaguid"; + // Unique per run: the SQL Server is shared by every checkout and CI, and a fixed name collided with a + // concurrent run's create / finally-drop. Digits only after "notaguid_", so it can never parse as a GUID. + var notAGuid = $"{AdminUiSqlServer.TestDatabasePrefix}20250101000000_notaguid_{Environment.ProcessId}_{Random.Shared.Next(1_000_000)}"; CreateDatabase(master, notAGuid);