From cef7d9b28198e2f00453864b482ce3788a8a8567 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 13 Jul 2026 12:07:00 -0400 Subject: [PATCH] refactor(opcuaserver): async-ify ServeRawPaged + HistoryReadEvents internals (03/S3) --- ...-ldap-historyread-async-plan.md.tasks.json | 2 +- .../OtOpcUaNodeManager.cs | 127 ++++++++++++------ 2 files changed, 86 insertions(+), 43 deletions(-) diff --git a/archreview/plans/R2-08-ldap-historyread-async-plan.md.tasks.json b/archreview/plans/R2-08-ldap-historyread-async-plan.md.tasks.json index 4b4da57f..a538f66f 100644 --- a/archreview/plans/R2-08-ldap-historyread-async-plan.md.tasks.json +++ b/archreview/plans/R2-08-ldap-historyread-async-plan.md.tasks.json @@ -10,7 +10,7 @@ { "id": "R2-08-T7", "subject": "S3 sequential-serving repro test (RED): MaxObservedConcurrency >= 2 on an 8-node Raw batch — fails on current code", "status": "completed", "blockedBy": [] }, { "id": "R2-08-T8", "subject": "ServerHistorianOptions: HistoryReadBatchParallelism/MaxConcurrentHistoryReadBatches/HistoryReadDeadline + Validate warnings + node-manager properties + hosted-service wiring", "status": "completed", "blockedBy": [] }, { "id": "R2-08-T9", "subject": "Async-ify ServeNode + Processed/AtTime arms (ct threading, OCE=>BadTimeout, behavior-neutral)", "status": "completed", "blockedBy": ["R2-08-T7", "R2-08-T8"] }, - { "id": "R2-08-T10", "subject": "Async-ify ServeRawPaged (page read + tie-cluster over-fetch) + extract ServeEventsAsync", "status": "pending", "blockedBy": ["R2-08-T9"] }, + { "id": "R2-08-T10", "subject": "Async-ify ServeRawPaged (page read + tie-cluster over-fetch) + extract ServeEventsAsync", "status": "completed", "blockedBy": ["R2-08-T9"] }, { "id": "R2-08-T11", "subject": "RunBounded fan-out in all four HistoryRead arms, single bridge per arm (turns T7 green) + upper-bound + isolation tests", "status": "pending", "blockedBy": ["R2-08-T9", "R2-08-T10"] }, { "id": "R2-08-T12", "subject": "Per-request HistoryRead deadline CTS — hung backend surfaces BadTimeout per node", "status": "pending", "blockedBy": ["R2-08-T11"] }, { "id": "R2-08-T13", "subject": "Process-wide concurrent-batch limiter — saturation degrades to BadTooManyOperations (lazy gate, bounded wait)", "status": "pending", "blockedBy": ["R2-08-T12"] }, diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 85499470..e6e8b0d4 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -1825,9 +1825,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 continue; } - ServeRawPaged( + // Interim: still bridged sequentially per node (bounded fan-out lands in a later slice). + ServeRawPagedAsync( handle, session, nodesToRead, results, errors, - details.StartTime, details.EndTime, details.NumValuesPerNode); + details.StartTime, details.EndTime, details.NumValuesPerNode, + CancellationToken.None).GetAwaiter().GetResult(); } } @@ -1922,41 +1924,75 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 continue; } - try - { - // NOT under the node-manager Lock — block-bridging the async source is safe here. - var sourceResult = _historianDataSource.ReadEventsAsync( - sourceName, - details.StartTime, - details.EndTime, - // NumValuesPerNode==0 means "no limit — return ALL values" per OPC UA - // Part 4/11, so translate it to UNBOUNDED (int.MaxValue) here. Passing the int<=0 - // backend-default-cap sentinel instead would silently truncate a "give me everything" - // events read at the backend default. A positive cap passes through (clamped). - EventMaxEvents(details.NumValuesPerNode), - CancellationToken.None).GetAwaiter().GetResult(); + // Interim: still bridged sequentially per node (bounded fan-out lands in a later slice). + ServeEventsAsync(handle, sourceName, details, selectClauses, results, errors, CancellationToken.None) + .GetAwaiter().GetResult(); + } + } - var historyEvent = ProjectEvents(sourceResult.Events, selectClauses); - results[handle.Index] = new SdkHistoryReadResult - { - // No events ⇒ GoodNoData (the notifier is historized, the window just held no events). - StatusCode = sourceResult.Events.Count == 0 ? StatusCodes.GoodNoData : StatusCodes.Good, - HistoryData = new ExtensionObject(historyEvent), - // We never issue continuation points — every read returns the full window in one shot. - ContinuationPoint = null, - }; - errors[handle.Index] = ServiceResult.Good; - } - catch (Exception ex) + /// + /// Serve one registered event-history notifier handle: reads the source's events over the request + /// window, projects them onto the select clauses, and fills the service-level results/errors slots. + /// Per-node error isolation matches — a backend throw becomes a Bad + /// status for THIS node only and never throws out of the batch; a deadline cancellation surfaces + /// BadTimeout. + /// + /// The notifier node handle; handle.Index indexes results/errors. + /// The registered historian event-source name for this notifier. + /// The event-read details (window + count cap). + /// The event-filter select clauses (the fields to emit, in order). + /// The service-level results list to fill at handle.Index. + /// The service-level errors list to fill at handle.Index. + /// Per-request deadline token. + private async Task ServeEventsAsync( + NodeHandle handle, + string? sourceName, + ReadEventDetails details, + SimpleAttributeOperandCollection selectClauses, + IList results, + IList errors, + CancellationToken ct) + { + try + { + // NOT under the node-manager Lock — awaiting the async source is safe here. + var sourceResult = await _historianDataSource.ReadEventsAsync( + sourceName, + details.StartTime, + details.EndTime, + // NumValuesPerNode==0 means "no limit — return ALL values" per OPC UA + // Part 4/11, so translate it to UNBOUNDED (int.MaxValue) here. Passing the int<=0 + // backend-default-cap sentinel instead would silently truncate a "give me everything" + // events read at the backend default. A positive cap passes through (clamped). + EventMaxEvents(details.NumValuesPerNode), + ct).ConfigureAwait(false); + + var historyEvent = ProjectEvents(sourceResult.Events, selectClauses); + results[handle.Index] = new SdkHistoryReadResult { - // One node's backend failure must not throw out of the batch — surface Bad for THIS node - // only. This manager carries no ILogger, so log via the SDK's static trace (see ServeNode). + // No events ⇒ GoodNoData (the notifier is historized, the window just held no events). + StatusCode = sourceResult.Events.Count == 0 ? StatusCodes.GoodNoData : StatusCodes.Good, + HistoryData = new ExtensionObject(historyEvent), + // We never issue continuation points — every read returns the full window in one shot. + ContinuationPoint = null, + }; + errors[handle.Index] = ServiceResult.Good; + } + catch (OperationCanceledException) + { + // The per-request deadline elapsed while reading this notifier — BadTimeout for THIS node only. + errors[handle.Index] = StatusCodes.BadTimeout; + results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTimeout }; + } + catch (Exception ex) + { + // One node's backend failure must not throw out of the batch — surface Bad for THIS node + // only. This manager carries no ILogger, so log via the SDK's static trace (see ServeNodeAsync). #pragma warning disable CS0618 // Type or member is obsolete - Utils.LogError(ex, "OtOpcUaNodeManager: HistoryReadEvents failed for node {0}", handle.NodeId); + Utils.LogError(ex, "OtOpcUaNodeManager: HistoryReadEvents failed for node {0}", handle.NodeId); #pragma warning restore CS0618 - errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; - results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported }; - } + errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; + results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported }; } } @@ -2149,7 +2185,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 /// The request window's (inclusive) lower bound, used for a fresh read. /// The (inclusive) upper bound of the read window; unchanged across pages. /// The client's per-page cap; 0 means "all values, no paging". - private void ServeRawPaged( + private async Task ServeRawPagedAsync( NodeHandle handle, ISession? session, IList nodesToRead, @@ -2157,7 +2193,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 IList errors, DateTime startTimeUtc, DateTime endUtc, - uint numValuesPerNode) + uint numValuesPerNode, + CancellationToken ct) { var inboundCp = nodesToRead[handle.Index].ContinuationPoint; @@ -2199,10 +2236,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 startUtc = startTimeUtc; } - // HistoryRead is NOT under the node-manager Lock — block-bridging the async source is safe. - var sourceResult = HistorianDataSource - .ReadRawAsync(tagname, startUtc, endUtc, numValuesPerNode, CancellationToken.None) - .GetAwaiter().GetResult(); + // HistoryRead is NOT under the node-manager Lock — awaiting the async source is safe. + var sourceResult = await HistorianDataSource + .ReadRawAsync(tagname, startUtc, endUtc, numValuesPerNode, ct) + .ConfigureAwait(false); var backendFull = HistoryPaging.IsFullPage(sourceResult.Samples.Count, numValuesPerNode); @@ -2232,9 +2269,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // Compute in uint so +1 can never wrap: Math.Max(1,…) also guards against a zero/negative // config value slipping through (Validate() warns but does not block startup). var overfetchCap = (uint)Math.Max(1, MaxTieClusterOverfetch) + 1u; - var cluster = HistorianDataSource - .ReadRawAsync(tagname, startUtc, startUtc, overfetchCap, CancellationToken.None) - .GetAwaiter().GetResult().Samples; + var cluster = (await HistorianDataSource + .ReadRawAsync(tagname, startUtc, startUtc, overfetchCap, ct) + .ConfigureAwait(false)).Samples; // Absurd burst: more ties than we're willing to buffer in memory. Preserve today's loud-fail // for that node rather than over-fetch an unbounded cluster; the operator's remedy is a @@ -2306,6 +2343,12 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 }; errors[handle.Index] = ServiceResult.Good; } + catch (OperationCanceledException) + { + // The per-request deadline elapsed while paging this node — BadTimeout for THIS node only. + errors[handle.Index] = StatusCodes.BadTimeout; + results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTimeout }; + } catch (Exception ex) { // One node's backend failure must not throw out of the batch — Bad for THIS node only.