Merge commit '2d75cecb' into codex/opcua2-cross-version
This commit is contained in:
Joseph Doherty
@@ -22,6 +22,37 @@ Design references: [docs/plans/2026-06-14-galaxy-phase-c-historian-design.md](pl
|
||||
|
||||
---
|
||||
|
||||
## SDK 2.0.1 compatibility (#782)
|
||||
|
||||
The server targets exact OPC Foundation SDK **2.0.1**, replacing **1.5.378.182**.
|
||||
The upgrade preserves existing application results, including the SQL backends'
|
||||
documented divergences from the old SDK below. Live `AggregateFilter` calculations
|
||||
use an application-private source port of the **37 legacy aggregate factories**,
|
||||
registered through the public SDK aggregate manager; they do not adopt SDK 2's new
|
||||
algorithms. Runtime and advertised aggregate defaults retain
|
||||
`TreatUncertainAsBad=false`. History backend capability sets remain separate from
|
||||
the live aggregate set; no backend gains aggregates or history operations through
|
||||
this upgrade.
|
||||
|
||||
The isolated [original-SDK oracle](../scripts/validation/opcua-aggregate-baseline/README.md)
|
||||
generates **6,144 vectors** using the published 1.5.378.182 package, covering value
|
||||
kinds, statuses, timestamps/picoseconds, reverse reads and edge configurations.
|
||||
These are expected-output fixtures, not a claim that release validation passed.
|
||||
The candidate never loads the old SDK binary. Older references below to "the SDK"
|
||||
as an aggregate oracle mean that preserved baseline, rather than SDK 2's algorithms.
|
||||
|
||||
SDK FILETIME-zero timestamps are translated back to the application's existing CLR
|
||||
unset-time sentinel at history boundaries. Opaque history continuation points remain
|
||||
session-owned, node/topology-bound and subject to current authorization; reconnect or
|
||||
rollback requires a fresh read. Request cancellation flows to the backend. No
|
||||
ConfigDb migration is planned; actual schema and serialization effects must be
|
||||
recorded at the [rollout gate](plans/2026-11-10-opcua-sdk-2-rollout.md).
|
||||
|
||||
`HistorianGateway.Client` and `HistorianGateway.Contracts` are already **0.4.0**
|
||||
(on gateway master `16111a5e`, historiangw#119 closed). This package alignment is
|
||||
separate from the real gateway validation tracked by historiangw#78; it is not
|
||||
live AVEVA or backend-conformance evidence for SDK 2.
|
||||
|
||||
## Historized TagConfig schema
|
||||
|
||||
A tag is historized by setting the **Historize this tag** checkbox and optional **Historian
|
||||
|
||||
@@ -1,185 +1,88 @@
|
||||
# OPC UA Client test fixture
|
||||
|
||||
Coverage map + gap inventory for the OPC UA Client (gateway / aggregation)
|
||||
driver.
|
||||
This is the fixture and command map for the driver/browser upgrade from SDK
|
||||
**1.5.378.182** to exact **2.0.1** (#782). It describes available tests; execution,
|
||||
counts, skips and release decisions belong in [#782](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/782)
|
||||
and the linked [execution evidence](../plans/2026-11-10-opcua-sdk-2-evidence.md).
|
||||
A listed test or historical run is not candidate release evidence.
|
||||
|
||||
**TL;DR:** Wire-level coverage now exists via
|
||||
[opc-plc](https://github.com/Azure-Samples/iot-edge-opc-plc) — Microsoft
|
||||
Industrial IoT's OPC UA PLC simulator running in Docker (task #215). Real
|
||||
Secure Channel, real Session, real MonitoredItem exchange against an
|
||||
independent server implementation. Unit tests still carry the exhaustive
|
||||
capability matrix (cert auth / security policies / reconnect / failover /
|
||||
attribute mapping). Gaps remaining: upstream-server-specific quirks
|
||||
(historian aggregates, typed ConditionType events, SDK-publish-queue edge
|
||||
behavior under load) — opc-plc uses the same OPCFoundation stack internally
|
||||
so fully-independent-stack coverage needs `open62541/open62541` as a second
|
||||
image (follow-up).
|
||||
## Fixture layers
|
||||
|
||||
## What the fixture is
|
||||
| Layer | Repository location | Scope and limitation |
|
||||
|---|---|---|
|
||||
| Driver unit | `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/` | Faked session contracts: lifecycle/generations, value conversion, per-item errors, policy selection, credentials, namespace remapping, subscription cleanup, discovery, alarms and host-owned reinitialization. No real secure channel. |
|
||||
| Browser unit | `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.Tests/` | Browse attributes/pagination and certificate-identity ownership, including failure cleanup. No external server. |
|
||||
| In-process wire | `tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/` | Real driver against OtOpcUa: `OpcUaClientDriverWireTests`, `OpcUaClientDriverStallWireTests`, `OpcUaClientDriverReconnectWireTests`. Real channels but both sides use this repository's SDK. |
|
||||
| External simulator | `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.IntegrationTests/` | `OpcUaClientSmokeTests`: real read, varied-type batch read and subscription against opc-plc. Does not prove all upstream products, secured authentication, write or alarm compatibility. |
|
||||
| Browser integration | `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.IntegrationTests/` | Separate fixture-backed browser tests; inspect fixture availability and individual results. |
|
||||
|
||||
**Integration layer** (task #215):
|
||||
`tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.IntegrationTests/` stands up
|
||||
`mcr.microsoft.com/iotedge/opc-plc:2.14.10` via `Docker/docker-compose.yml`
|
||||
on `opc.tcp://10.100.0.35:50000` (the shared Docker host; override via
|
||||
`OPCUA_SIM_ENDPOINT`). `OpcPlcFixture` probes the port at
|
||||
collection init + skips tests with a clear message when the container's
|
||||
not running (matches the Modbus/pymodbus + S7/python-snap7 skip pattern).
|
||||
Docker is the launcher — no PowerShell wrapper needed because opc-plc
|
||||
ships pre-containerized. Compose-file flags: `--ut` (unsecured transport
|
||||
advertised), `--aa` (auto-accept client certs — opc-plc's cert trust store
|
||||
resets on each spin-up), `--alm` (alarm simulation for IAlarmSource
|
||||
follow-up coverage), `--pn=50000` (port).
|
||||
The SDK 2 unit boundary cases include `OpcUaClientSdkValueBoundaryTests` and
|
||||
`OpcUaClientContinuationPointReleaseTests`. They preserve CLR value types and
|
||||
null/empty buffers, reject unsupported matrix items without shifting batch results,
|
||||
and release abandoned browse points. Certificate tests cover lifetime through
|
||||
session close; unit mocks alone do not establish trust-chain interoperability.
|
||||
|
||||
**Unit layer**:
|
||||
`tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/` is still the primary
|
||||
coverage. Tests inject fakes through the driver's construction path; the
|
||||
OPCFoundation.NetStandard `Session` surface is wrapped behind an interface
|
||||
the tests mock.
|
||||
The in-process stall proxy holds bytes without closing the session, then resumes;
|
||||
its hard-close case refuses connections. These exercise stale notification rejection,
|
||||
quality transitions and explicit host reinitialization. The driver does not use
|
||||
`SessionReconnectHandler` or `TransferSubscriptions`; historical transfer-based
|
||||
coverage descriptions do not apply to the current lifecycle (#670/#676).
|
||||
|
||||
## What it actually covers
|
||||
The driver has **no upstream HistoryRead capability** (#655). Its tags can be
|
||||
recorded into the local server-wide historian; backend conformance and aggregate
|
||||
oracles are documented in [Historian.md](../Historian.md).
|
||||
|
||||
### Integration (opc-plc Docker, task #215)
|
||||
## Running locally
|
||||
|
||||
- `OpcUaClientSmokeTests.Client_connects_and_reads_StepUp_node_through_real_OPC_UA_stack` —
|
||||
full Secure Channel + Session + `ns=3;s=StepUp` Read round-trip
|
||||
- `OpcUaClientSmokeTests.Client_reads_batch_of_varied_types_from_live_simulator` —
|
||||
batch Read of UInt32 / Int32 / Boolean; asserts `bool`-specific Variant
|
||||
decoding to catch a common attribute-mapping regression
|
||||
- `OpcUaClientSmokeTests.Client_subscribe_receives_StepUp_data_changes_from_live_server` —
|
||||
real `MonitoredItem` subscription against `ns=3;s=FastUInt1` (ticks every
|
||||
100 ms); asserts `OnDataChange` fires within 3 s of subscribe
|
||||
Run from the repository root. Build first and use a fresh result directory for
|
||||
**each** project. The [runbook](../plans/2026-11-10-opcua-sdk-2-rollout.md#verification-commands)
|
||||
provides the result helper that preserves test exit status and rejects all-skipped
|
||||
runs. For example, after defining that helper:
|
||||
|
||||
Wire-level surfaces verified: `IDriver` + `IReadable` + `ISubscribable` +
|
||||
`IHostConnectivityProbe` (via the Secure Channel exchange).
|
||||
```bash
|
||||
opc_test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests.csproj
|
||||
opc_test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.Tests/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser.Tests.csproj
|
||||
opc_test tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.csproj
|
||||
```
|
||||
|
||||
### In-process wire tests (no opc-plc)
|
||||
The simulator compose file is
|
||||
`tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.IntegrationTests/Docker/docker-compose.yml`.
|
||||
It launches `mcr.microsoft.com/iotedge/opc-plc:2.14.10` with `--ut --aa --alm --pn=50000`.
|
||||
Use the shared-host procedure in [infra/README.md](../../infra/README.md); coordinate
|
||||
shared fixture use before restarting containers. These dev flags do not establish
|
||||
production trust or secure-policy coverage.
|
||||
|
||||
`tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverWireTests` drives the real driver
|
||||
against the in-process OtOpcUa server (`ServerEventRig`, authenticating with the rig's API-key credential):
|
||||
read (+ missing node), write through the node-write gateway (+ read-only refusal, + failed-write revert),
|
||||
subscribe/unsubscribe, host-status Running -> not-Running when the server stops, and discovery. These need
|
||||
no Docker. There is no HistoryRead leg: the driver has no history capability (#655).
|
||||
`OpcPlcFixture` probes `OPCUA_SIM_ENDPOINT` (default `opc.tcp://10.100.0.35:50000`)
|
||||
and skips when unavailable. With an available, reserved fixture:
|
||||
|
||||
### Unit
|
||||
```bash
|
||||
export OPCUA_SIM_ENDPOINT=opc.tcp://10.100.0.35:50000
|
||||
opc_test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.IntegrationTests.csproj
|
||||
```
|
||||
|
||||
The surface is broad because `OpcUaClientDriver` is the richest-capability
|
||||
driver in the fleet (it's a gateway for another OPC UA server, so it
|
||||
mirrors the full capability matrix):
|
||||
A nonzero suite count does not prove every required scenario ran: inspect individual
|
||||
TRX results and record required skips separately. No tests were run by preparing
|
||||
this documentation.
|
||||
|
||||
- `OpcUaClientDriverScaffoldTests` — `IDriver` lifecycle
|
||||
- `OpcUaClientReadWriteTests` — read + write lifecycle
|
||||
- `OpcUaClientSubscribeAndProbeTests` — monitored-item subscription + probe
|
||||
state transitions
|
||||
- `OpcUaClientDiscoveryTests` — `GetEndpoints` + endpoint selection
|
||||
- `OpcUaClientAttributeMappingTests` — OPC UA node attribute → driver value
|
||||
mapping
|
||||
- `OpcUaClientSecurityPolicyTests` — `SignAndEncrypt` / `Sign` / `None`
|
||||
policy negotiation contract
|
||||
- `OpcUaClientCertAuthTests` — cert store paths, revocation-list config
|
||||
- `OpcUaClientReconnectTests` — SDK reconnect hook + `TransferSubscriptions`
|
||||
across the disconnect boundary
|
||||
- `OpcUaClientFailoverTests` — primary → secondary session fallback per
|
||||
driver config
|
||||
- `OpcUaClientAlarmTests` — A&E severity bucket (1–1000 → Low / Medium /
|
||||
High / Critical), subscribe / unsubscribe / ack contract
|
||||
## Required compatibility evidence
|
||||
|
||||
Capability surfaces whose contract is verified: `IDriver`, `ITagDiscovery`,
|
||||
`IReadable`, `IWritable`, `ISubscribable`, `IHostConnectivityProbe`,
|
||||
`IAlarmSource`.
|
||||
Release validation must identify both application builds and SDK versions, endpoint
|
||||
policy, token kind, certificate chain, configuration and executed capabilities.
|
||||
Use the baseline client against the candidate server and candidate client against
|
||||
the baseline server, plus the actual ScadaBridge adapter and external upstreams.
|
||||
For secured paths include trusted SignAndEncrypt, API-key allow/deny/revocation,
|
||||
certificate rejection and restart/rebind. Apply only operations the peer implements;
|
||||
a simulator need not implement OtOpcUa's API-key or historian contract.
|
||||
|
||||
## What it does NOT cover
|
||||
For alarms verify typed EventFilter select clauses, result-field indexes, actual
|
||||
ConditionType/SourceNode/EventId identity, refresh, acknowledgements and no replay
|
||||
after an ambiguous transport failure. For all supported paths preserve existing
|
||||
operation statuses, locale behavior and null/empty filter meaning. Constructing an
|
||||
empty SDK 2 `ContentFilter` is an API adaptation, not evidence of a changed filter
|
||||
contract. Record any stricter protocol response and consumer impact before acceptance.
|
||||
|
||||
### 1. Full real-stack exchange (unit tests only)
|
||||
|
||||
The **unit** suite mocks `Session.ReadAsync`, `Session.CreateSubscription`,
|
||||
`Session.AddItem`, etc. — no UA Secure Channel is opened. The **integration**
|
||||
suite (`OpcUaClientSmokeTests`, task #215) does open a real Secure Channel
|
||||
against opc-plc and exercises Read + Subscribe end-to-end. What remains
|
||||
untested even in the integration suite: certificate validation under
|
||||
non-anonymous security policies, signing/encryption, nonce handling, chunk
|
||||
assembly, keep-alive cadence — all SDK-internal.
|
||||
|
||||
### 2. Subscription transfer across reconnect
|
||||
|
||||
Contract test: "after a simulated reconnect, `TransferSubscriptions` is
|
||||
called with the right handles." Real behavior: SDK re-publishes against the
|
||||
new channel and some events can be lost depending on publish-queue state.
|
||||
The lossy window is not characterized.
|
||||
|
||||
### 3. Large-scale subscription stress
|
||||
|
||||
100+ monitored items with heterogeneous publish intervals under a single
|
||||
session — the shape that breaks publish-queue-size tuning in the wild — is
|
||||
not exercised.
|
||||
|
||||
### 4. Real A&E events
|
||||
|
||||
Alarm subscription is mocked via filtered monitored items; the actual
|
||||
`EventFilter` select-clause behavior against a server that exposes typed
|
||||
ConditionType events (non-base `BaseEventType`) is not verified.
|
||||
|
||||
### 6. Authentication variants
|
||||
|
||||
- Anonymous, UserName/Password, X509 cert tokens — each is contract-tested
|
||||
but not exchanged against a server that actually enforces each.
|
||||
- LDAP-backed `UserName` (matching this repo's server-side
|
||||
`LdapOpcUaUserAuthenticator`) requires a live LDAP round-trip; not tested.
|
||||
|
||||
## When to trust OpcUaClient tests, when to reach for a server
|
||||
|
||||
| Question | Unit tests | Integration (opc-plc) | Real upstream server |
|
||||
| --- | --- | --- | --- |
|
||||
| "Does severity 750 bucket as High?" | yes | - | yes |
|
||||
| "Does the driver call `TransferSubscriptions` after reconnect?" | yes | - | yes |
|
||||
| "Does a real OPC UA read round-trip work?" | no | yes | yes |
|
||||
| "Does a real OPC UA subscribe deliver changes?" | no | yes | yes |
|
||||
| "Does write round-trip work against a live server?" | no | no (not yet exercised) | yes (required) |
|
||||
| "Does event-filter-based alarm subscription return ConditionType events?" | no | no | yes (required) |
|
||||
| "Does history read from AVEVA Historian return correct aggregates?" | no | no | yes (required) |
|
||||
| "Does the SDK's publish queue lose notifications under load?" | no | no | yes (stress) |
|
||||
|
||||
## Follow-up candidates
|
||||
|
||||
The easiest win here is to **wire the client driver tests against this
|
||||
repo's own server**. The v2 integration test project
|
||||
`tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/DualEndpointTests.cs`
|
||||
(the v2 replacement for the retired v1 `OpcUaServerIntegrationTests`) already
|
||||
stands up a real OPC UA server on a non-default port with a seeded
|
||||
FakeDriver. An `OpcUaClientLiveLoopbackTests` that connects the client
|
||||
driver to that server would give:
|
||||
|
||||
- Real Secure Channel negotiation
|
||||
- Real Session / Subscription / MonitoredItem exchange
|
||||
- Real read/write round-trip
|
||||
- Real certificate validation (the integration test already sets up PKI)
|
||||
|
||||
It wouldn't cover upstream-server-specific quirks (AVEVA Historian, Kepware,
|
||||
Prosys), but it would cover 80% of the SDK surface the driver sits on top
|
||||
of.
|
||||
|
||||
Beyond that:
|
||||
|
||||
1. **Prosys OPC UA Simulation Server** — free, Windows-available, scriptable.
|
||||
2. **UaExpert Server-Side Simulator** — Unified Automation's sample server;
|
||||
good coverage of typed ConditionType events.
|
||||
3. **Dedicated historian integration lab** — only path for
|
||||
historian-specific coverage.
|
||||
|
||||
## Key fixture / config files
|
||||
|
||||
- `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/` — unit tests with
|
||||
mocked `Session`
|
||||
- `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.IntegrationTests/OpcPlcFixture.cs`
|
||||
— collection fixture; parses `OPCUA_SIM_ENDPOINT` (default
|
||||
`opc.tcp://10.100.0.35:50000`), TCP-probes at collection init, records
|
||||
`SkipReason` when unreachable
|
||||
- `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.IntegrationTests/OpcUaClientSmokeTests.cs`
|
||||
— wire-level test suite (3 `[Fact]` methods: read, batch read, subscribe)
|
||||
- `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.IntegrationTests/Docker/docker-compose.yml`
|
||||
— `mcr.microsoft.com/iotedge/opc-plc:2.14.10` with `--ut --aa --alm --pn=50000`
|
||||
- `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs` — ctor +
|
||||
session-factory seam tests mock through; implements `IAlarmSource`;
|
||||
does NOT implement `IRediscoverable`
|
||||
- `tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/DualEndpointTests.cs` —
|
||||
the v2 dual-endpoint integration harness a future loopback client test could
|
||||
piggyback on (v1 `OpcUaServerIntegrationTests.cs` retired with the v1 server project)
|
||||
opc-plc uses the OPC Foundation stack too. Independent upstream behavior, secured
|
||||
authentication variants, large publish queues, real alarm products, packet-loss
|
||||
windows and performance require separate evidence. The rollout runbook links the
|
||||
pending gates; no simulator run substitutes for ScadaBridge, live gateway, Windows,
|
||||
Linux or mixed-version cluster validation.
|
||||
|
||||
@@ -9,8 +9,9 @@ and for the test-harness map read [OpcUaClient-Test-Fixture.md](OpcUaClient-Test
|
||||
A **remote OPC UA server**. This driver runs the *opposite* direction from the
|
||||
usual "server exposes PLC data" flow: it acts as an OPC UA **client**, opens a
|
||||
`Session` against an upstream server, and re-exposes that server's address space
|
||||
through the local OtOpcUa server. Browse, read, write, subscribe, alarm, and
|
||||
history calls are passed through to the upstream endpoint.
|
||||
through the local OtOpcUa server. Browse, read, write, subscribe and alarm
|
||||
operations use the upstream endpoint. HistoryRead uses the local server-wide
|
||||
historian instead; see [History](#history-not-supported-gitea-655).
|
||||
|
||||
It is built on the OPC Foundation UA .NET Standard reference SDK and runs
|
||||
in-process in the OtOpcUa server's .NET 10 AnyCPU host — pure managed, no
|
||||
@@ -20,6 +21,31 @@ out-of-process isolation.
|
||||
> a remote OPC UA endpoint by hand, point the general-purpose
|
||||
> [Client CLI](../Client.CLI.md) at it directly.
|
||||
|
||||
## SDK 2.0.1 compatibility (#782)
|
||||
|
||||
The driver, shared session layer and AdminUI browser target exact SDK **2.0.1**
|
||||
(previous **1.5.378.182**). Their existing abstractions and host-owned reconnect
|
||||
lifecycle remain in place. SDK immutable values are detached at the boundary into
|
||||
existing CLR scalars and one-dimensional arrays, preserving null/empty ByteStrings,
|
||||
UTC timestamps, Guid values, numeric widths, item order and per-item status. Matrix
|
||||
payloads remain unsupported and fail per item; this upgrade adds no matrix or
|
||||
upstream-history capability.
|
||||
|
||||
Certificate-based identities retain their loaded certificate until their owning
|
||||
session closes, including failure cleanup. The browser releases abandoned browse
|
||||
continuation points. Endpoint policy selection, namespace-URI remapping, old-session
|
||||
notification rejection and the existing refusal to replay non-idempotent writes or
|
||||
alarm acknowledgements remain compatibility requirements.
|
||||
|
||||
No application status-code, locale-selection or event-filter semantic change is
|
||||
approved by this upgrade. SDK value wrappers and empty-filter construction are API
|
||||
adaptations; callers must continue to inspect operation and select-clause results
|
||||
by index. Confirm actual client/server combinations with the
|
||||
[fixture map](OpcUaClient-Test-Fixture.md) and
|
||||
[rollout gates](../plans/2026-11-10-opcua-sdk-2-rollout.md) before deployment.
|
||||
Historical SDK 1.5 observations elsewhere on this page describe their original
|
||||
measurements; they are not fresh SDK 2.0.1 interoperability evidence.
|
||||
|
||||
## Project split
|
||||
|
||||
| Project | Target | Role |
|
||||
@@ -365,7 +391,7 @@ They are not part of `DriverHealth` or the
|
||||
## Testing
|
||||
|
||||
- **Unit tests** — `tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/`
|
||||
cover the session lifecycle, namespace remapping, alarm/history passthrough,
|
||||
cover the session lifecycle, namespace remapping, alarm dispatch,
|
||||
and config binding against a faked SDK session.
|
||||
- **Integration fixture** — exercises the driver against a reference OPC UA
|
||||
server (opc-plc) on the shared docker host; see
|
||||
|
||||
@@ -0,0 +1,284 @@
|
||||
# OPC UA SDK 2.0.1 rollout and rollback
|
||||
|
||||
This operator runbook accompanies [#782](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/782),
|
||||
the [implementation plan](2026-11-10-opcua-sdk-2-upgrade.md) and
|
||||
[execution evidence](2026-11-10-opcua-sdk-2-evidence.md). Gitea is the single
|
||||
status system. This draft was checked against source `a21104d2`; it does not
|
||||
approve an artifact or record a deployment. The coordinator must attach final
|
||||
commit-specific evidence in #782 before advancing a gate. All placeholders below
|
||||
are **pending evidence**, not passing results.
|
||||
|
||||
The target is exact OPC Foundation SDK **2.0.1**; the known-good SDK baseline is
|
||||
**1.5.378.182** (application baseline `83ed880e4471344deaf008da9bd028c7e170bdc7`).
|
||||
Restore the actual archived known-good application as a unit during rollback.
|
||||
Package downgrades alone cannot reverse the application/API migration.
|
||||
|
||||
## Operating constraints
|
||||
|
||||
**The remote-execution hold on wonder boxes vd03/vq03/vq04 remains binding.**
|
||||
No WinRM, psbridge or replacement remote-exec channel may be used until the user
|
||||
clears a channel. UNC/SMB file copying and permitted SQL access do not authorize
|
||||
service control, PowerShell execution or deployment on those hosts. Steps requiring
|
||||
a command on a held host remain pending. No deployment was performed for this draft.
|
||||
|
||||
Run the service commands below only from an approved local operator session on an
|
||||
authorized Windows node. Record the node, account, channel and maintenance window
|
||||
in #782 first. Linux runtime/service handling requires its own approved environment
|
||||
and smoke record; this runbook supplies no invented Linux service-unit name.
|
||||
Do not use `scripts/install/Refresh-Services.ps1` as a blanket upgrade action: it
|
||||
also touches gateway/retired sidecar paths outside this SDK rollout.
|
||||
|
||||
## Preserved contract
|
||||
|
||||
- Existing raw/UNS namespace URIs, identifier forms, endpoint profiles, API-key/anonymous
|
||||
admission, ACL flag behavior, role and key revocation gates remain the contract.
|
||||
Source uses public transport peer evidence for miss budgets, not an IP admission rule.
|
||||
- Scalar leniency, checked numeric-array conversion and pass-through driver behavior
|
||||
stay as before. SDK immutable values are detached at boundaries; no new matrix support.
|
||||
- No application status-code, locale or event-filter semantic change is approved.
|
||||
SDK status wrappers and explicit empty `ContentFilter` construction are API changes,
|
||||
not evidence of changed protocol values. Record any stricter response, field-level
|
||||
filter error or localization difference with a consumer-impact decision before go.
|
||||
- Auditing stays opt-in. Ordinary SDK automatic audit delivery retains its type
|
||||
restrictions. Only the application's rejected-write audit gets a private, temporary
|
||||
receive-permission context after current source `Read` ACL and audit/encryption checks;
|
||||
no shared role metadata or general audit access is widened. See [security](../security.md).
|
||||
- The private source port retains the **37 legacy aggregate factories** and existing
|
||||
application results rather than adopting SDK 2 algorithms. Original-SDK fixtures
|
||||
contain **6,144 vectors**. `TreatUncertainAsBad=false` stays explicit; historian
|
||||
backend aggregate sets and existing documented differences remain unchanged.
|
||||
SDK FILETIME-zero maps to the existing CLR unset-time sentinel at history boundaries.
|
||||
- No new historian backend, companion specification, PubSub, ManagedSession, AOT or
|
||||
cluster rolling-upgrade capability is claimed. HistorianGateway Client/Contracts
|
||||
**0.4.0** is already on gateway master `16111a5e` (historiangw#119 closed);
|
||||
the real gateway gate historiangw#78 remains separate from package alignment.
|
||||
|
||||
## Release record and preflight
|
||||
|
||||
Fill this record in #782 with links to immutable logs/manifests; do not paste keys,
|
||||
passwords, connection strings, private certificates or unredacted environment dumps.
|
||||
|
||||
| Required evidence | Record to supply before release |
|
||||
|---|---|
|
||||
| Candidate source and graph | Pending: full commit, clean/dirty state, SDK/runtime/OS, resolved package graph showing exact 2.0.1 |
|
||||
| Known-good recovery set | Pending: deployed application version, full binary/config archive SHA-256, protected PKI backup location and permission record |
|
||||
| Candidate artifacts | Pending: Windows and Linux publish directories/archives, RID/self-contained choice, SHA-256 manifest, dependencies and script-worker contents |
|
||||
| G2 compilation | Pending: restore/build logs and warning/error counts at the final SHA |
|
||||
| G3 behavior/security | Pending: project TRX/counts, required case results, positive and negative authorization, audit privacy, CP lifecycle and 6,144-vector comparison |
|
||||
| Interoperability | Pending: baseline/candidate client-server pairs, actual ScadaBridge adapter, upstream driver/browser products and mixed-version cluster gate |
|
||||
| Historian | Pending: configured SqlServer/QuestDB/Gateway capability-specific results; historiangw#78 link/evidence for live gateway |
|
||||
| G4 resources/performance | Pending: baseline comparison, at least three measured runs, two-hour soak, Windows/Linux start-stop-twice smoke with preserved PKI |
|
||||
| Review/operations | Pending: independent security/release review, authorized channel/window, named operator, canary/rollback results and decision |
|
||||
|
||||
Before staging:
|
||||
|
||||
1. Inventory actual service executable/arguments, roles, endpoints, advertised host,
|
||||
application URI, raw/UNS namespace identities, certificate thumbprints, active/standby
|
||||
state and connected client builds. Record configured queues, request/transport quotas,
|
||||
history caps and expected load. Establish performance budgets and stop thresholds.
|
||||
2. Archive complete known-good binaries and node configuration, preserving access controls.
|
||||
Securely back up PKI own/private keys, peer/issuer chains, CRLs and their permissions.
|
||||
Verify recovery access under the service identity, private-key presence, expiry,
|
||||
application URI/hostname match and CA/key-usage suitability. Retain existing certificates.
|
||||
3. Inventory ConfigDb, LocalDb, historian/outbox and secret-provider state/locations. No
|
||||
ConfigDb migration is planned. Record the actual schema/serialization diff, including
|
||||
Akka/deployment contracts; if persistence changes appear, reopen the design gate and
|
||||
define data rollback before deployment. Never overwrite live historian/outbox data
|
||||
with an old binary backup or rewrite persisted sessions/subscriptions.
|
||||
4. Publish into fresh directories, hash artifacts, stage separately from the live install
|
||||
and validate graph contents. Keep baseline and candidate ports, PKI, config and stores
|
||||
separate in the isolated rig; never run both against the same mutable fixture.
|
||||
5. Confirm required positive/negative checks, reviewers and recovery artifact access.
|
||||
Missing evidence, inaccessible environments or an uncleared command channel means hold.
|
||||
|
||||
## Verification commands
|
||||
|
||||
These commands are recipes, not execution results. Run from the repository root in
|
||||
Bash with .NET 10. Record the final SHA and logs in a fresh evidence directory; inspect
|
||||
exit status before continuing. Do not pipe a failing build/test into a successful logger.
|
||||
|
||||
```bash
|
||||
git rev-parse HEAD
|
||||
git status --short
|
||||
dotnet --info
|
||||
dotnet restore ZB.MOM.WW.OtOpcUa.slnx
|
||||
dotnet build ZB.MOM.WW.OtOpcUa.slnx -c Release --no-restore
|
||||
```
|
||||
|
||||
Define the helper once after a successful build. Each invocation creates its own
|
||||
fresh TRX directory. It preserves both test failure and zero-executed-test failure.
|
||||
A successful guard still requires inspection of required cases and skips.
|
||||
|
||||
```bash
|
||||
opc_test() {
|
||||
local opc_project="$1"
|
||||
local opc_results
|
||||
opc_results=$(mktemp -d "${TMPDIR:-/tmp}/opcua2-test.XXXXXXXX") || return
|
||||
local opc_test_status=0
|
||||
dotnet test "$opc_project" -c Release --no-build --logger trx \
|
||||
--results-directory "$opc_results" || opc_test_status=$?
|
||||
local opc_count_status=0
|
||||
MIN_EXECUTED=1 bash scripts/ci/assert-not-all-skipped.sh "$opc_results"/*.trx \
|
||||
|| opc_count_status=$?
|
||||
printf 'TRX directory: %s\n' "$opc_results"
|
||||
if [ "$opc_test_status" -ne 0 ]; then return "$opc_test_status"; fi
|
||||
return "$opc_count_status"
|
||||
}
|
||||
opc_test tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.csproj
|
||||
opc_test tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.csproj
|
||||
opc_test tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests.csproj
|
||||
opc_test tests/Client/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests/ZB.MOM.WW.OtOpcUa.Client.Shared.Tests.csproj
|
||||
opc_test tests/Client/ZB.MOM.WW.OtOpcUa.Client.CLI.Tests/ZB.MOM.WW.OtOpcUa.Client.CLI.Tests.csproj
|
||||
```
|
||||
|
||||
Driver/browser and simulator commands are in the
|
||||
[fixture map](../drivers/OpcUaClient-Test-Fixture.md#running-locally). Backend and
|
||||
live historian prerequisites are in [Historian.md](../Historian.md) and
|
||||
[infra/README.md](../../infra/README.md). Live tests can write sandbox data: use
|
||||
only designated sandbox tags and approved credentials through the existing secret
|
||||
provider/environment. A skipped live/DB fixture never satisfies its gate.
|
||||
Generate original aggregate evidence only with the isolated
|
||||
[oracle procedure](../../scripts/validation/opcua-aggregate-baseline/README.md),
|
||||
a new output path and original 1.5.378.182 binary; never derive expectations from
|
||||
the candidate's calculators.
|
||||
|
||||
Also run the full CI-equivalent test leg after integration; it does not replace
|
||||
fixture-specific or live gates:
|
||||
|
||||
```bash
|
||||
opc_solution_test() {
|
||||
local opc_results
|
||||
opc_results=$(mktemp -d "${TMPDIR:-/tmp}/opcua2-solution.XXXXXXXX") || return
|
||||
local opc_test_status=0
|
||||
dotnet test ZB.MOM.WW.OtOpcUa.slnx -c Release --no-build \
|
||||
-m:1 -p:TestTfmsInParallel=false --settings scripts/ci/ci.runsettings \
|
||||
--filter 'Category!=E2E&Category!=LiveIntegration' \
|
||||
--logger trx --results-directory "$opc_results" || opc_test_status=$?
|
||||
local opc_count_status=0
|
||||
MIN_EXECUTED=12000 bash scripts/ci/assert-not-all-skipped.sh "$opc_results"/*.trx \
|
||||
|| opc_count_status=$?
|
||||
printf 'TRX directory: %s\n' "$opc_results"
|
||||
if [ "$opc_test_status" -ne 0 ]; then return "$opc_test_status"; fi
|
||||
return "$opc_count_status"
|
||||
}
|
||||
opc_solution_test
|
||||
```
|
||||
|
||||
Publish the Host project to a fresh operator-selected `$OPC_PUBLISH_DIR` using the
|
||||
approved `$OPC_RID` and deployment's self-contained choice. The Windows install
|
||||
expects `OtOpcUa.Host.exe`; [ServiceHosting.md](../ServiceHosting.md) also describes
|
||||
script-worker packaging. Example self-contained Windows artifact command:
|
||||
|
||||
```bash
|
||||
dotnet publish src/Server/ZB.MOM.WW.OtOpcUa.Host/ZB.MOM.WW.OtOpcUa.Host.csproj \
|
||||
-c Release -r win-x64 --self-contained true -o "$OPC_PUBLISH_DIR"
|
||||
```
|
||||
|
||||
Repeat with the approved Linux RID/framework settings in its validation environment;
|
||||
cross-publishing is not runtime/install evidence. Hash the complete staged artifact
|
||||
and retain a manifest outside it (PowerShell 7, where `$Candidate` and `$Manifest`
|
||||
are approved absolute local paths and the manifest is outside the candidate):
|
||||
|
||||
```powershell
|
||||
Get-ChildItem -LiteralPath $Candidate -Recurse -File |
|
||||
Get-FileHash -Algorithm SHA256 |
|
||||
Export-Csv -LiteralPath $Manifest -NoTypeInformation
|
||||
```
|
||||
|
||||
## Sequence and stop/go decision
|
||||
|
||||
Progress only after the preceding stage's recorded checks and explicit gate decision:
|
||||
**isolated rig → ScadaBridge/upstream interoperability → DEV canary → QA pair →
|
||||
production canary**. Attach source/artifact identity, start/end times, operator,
|
||||
checks, observations and go/hold/rollback decision for each stage to #782.
|
||||
|
||||
A standby-first pair restart is allowed only after the application's mixed-version
|
||||
Akka serialization, deployment/config and failover gate succeeds. Verify standby
|
||||
health, security, data, history and events; exercise failover before upgrading its
|
||||
peer. OPC UA wire interoperability alone does not prove cluster rolling safety.
|
||||
Without that gate, arrange a maintenance window and coordinated pair restart;
|
||||
record expected service interruption and recovery checks.
|
||||
|
||||
The following Windows commands are **local-session steps only**. Bind `$Install`
|
||||
to the inspected current binary directory, `$Candidate` to the reviewed staged
|
||||
artifact and `$Backup` to a new protected directory. Keep PKI/data stores outside
|
||||
binary replacement; if they are inside the install tree, define exclusions and a
|
||||
separate protected backup before using these commands. Save configuration outside
|
||||
`$Candidate`, then apply the reviewed node configuration to the staging directory;
|
||||
never overwrite it with publisher defaults or expose secrets in command output.
|
||||
|
||||
```powershell
|
||||
Get-CimInstance Win32_Service -Filter "Name='OtOpcUaHost'" |
|
||||
Select-Object Name, State, PathName, StartName
|
||||
Stop-Service -Name OtOpcUaHost -ErrorAction Stop
|
||||
(Get-Service -Name OtOpcUaHost).WaitForStatus('Stopped', [TimeSpan]::FromSeconds(60))
|
||||
robocopy $Install $Backup /E /COPYALL /DCOPY:DAT /R:1 /W:1
|
||||
if ($LASTEXITCODE -ge 8) { throw 'Known-good backup failed' }
|
||||
# After checking the backup and candidate manifest, swap complete binary directories.
|
||||
# Use sibling paths on the same volume; the service PathName must still resolve to $Install.
|
||||
Move-Item -LiteralPath $Install -Destination $PreviousInstall -ErrorAction Stop
|
||||
Move-Item -LiteralPath $Candidate -Destination $Install -ErrorAction Stop
|
||||
Start-Service -Name OtOpcUaHost -ErrorAction Stop
|
||||
(Get-Service -Name OtOpcUaHost).WaitForStatus('Running', [TimeSpan]::FromSeconds(60))
|
||||
```
|
||||
|
||||
`$PreviousInstall` must be a fresh reviewed sibling path. Handle any failure between
|
||||
moves by restoring that complete directory before starting. Service `Running` is
|
||||
only a process check. Run the substantive checks below after every start and failover:
|
||||
|
||||
- Check `/healthz`, `/health/ready` and role-appropriate `/health/active` at the
|
||||
recorded HTTP base URL; compare ServiceLevel and active/standby state to baseline.
|
||||
- Confirm certificate thumbprints, trust/issuer chains, endpoint policies/application
|
||||
URI and namespace-URI/NodeId identity. Prove trusted SignAndEncrypt works and
|
||||
untrusted/revoked/inappropriate certificates and wrong keys fail without downgrade.
|
||||
- Exercise raw and UNS browse/read, scalar/array writes on sandbox nodes, subscriptions,
|
||||
quality and source/server timestamps. Verify current legitimate ACL allows as well
|
||||
as denies, key expiry/revocation and audience-specific event/audit filtering.
|
||||
- Verify alarms/refresh/ack and no missing or duplicate events; verify configured raw,
|
||||
modified, processed and event history, paging/release/reauthentication and expected
|
||||
unsupported operations. Verify recorder/outbox progress without duplicate replay.
|
||||
- Observe reconnects, stale values, legitimate-call denials, continuation-point errors,
|
||||
session/item/certificate-handle growth, shutdown, queue depths and latency.
|
||||
|
||||
Use [Client.CLI.md](../Client.CLI.md) for exact connect/browse/read/subscribe/history
|
||||
and alarm syntax. Supply credentials through the approved operator mechanism; do
|
||||
not record command lines containing API-key passwords in release evidence.
|
||||
|
||||
Stop promotion and restore the known-good application for any authorization regression,
|
||||
missing/duplicate alarm or history data, changed identifiers, insecure fallback,
|
||||
restart failure or sustained performance-budget miss. Compare like hardware/config/load:
|
||||
at least three warmed measurements, provisional limits ≤10% median throughput loss
|
||||
and ≤15% p95 latency/allocation/RSS growth, plus at least two hours of lifecycle soak.
|
||||
Any deviation needs investigation and a recorded acceptance decision. Do not lower
|
||||
thresholds or count absent environments as passed.
|
||||
|
||||
## Rollback
|
||||
|
||||
1. Freeze promotion; record the trigger and affected nodes/clients in #782. Preserve
|
||||
candidate logs and manifests with secrets redacted. Stop the candidate through the
|
||||
approved local channel; the remote-exec hold still applies during rollback.
|
||||
2. With the service stopped, move the candidate install to a fresh quarantine path,
|
||||
restore the complete `$PreviousInstall` (or verified `$Backup`) to the inspected
|
||||
service path, and restore the matching node configuration and preserved PKI with
|
||||
original permissions. Check hashes/thumbprints before starting. Do not delete or
|
||||
roll back live historian, outbox or ConfigDb data without a reviewed data strategy.
|
||||
3. Start `OtOpcUaHost` locally with the same `Start-Service`/wait commands above.
|
||||
On Linux, use only the environment's approved recorded supervisor procedure.
|
||||
Verify health, endpoint/namespace/certificate identity and active/standby state.
|
||||
4. Reconnect clients as needed. In-memory sessions, subscriptions and continuation
|
||||
points do not survive process rollback; discard old points and begin new reads.
|
||||
Rebind namespaces by URI. Never replay ambiguous non-idempotent writes or alarm
|
||||
acknowledgements. Confirm API-key writes/ack, subscriptions, alarm counts and history
|
||||
continuity with the same positive and negative checks used at rollout.
|
||||
5. Record recovery time, checks, data reconciliation and final hold/go decision.
|
||||
If rolling source back, revert package and application commits together; never
|
||||
mix old package binaries with SDK 2 application assemblies. Keep candidate evidence
|
||||
for diagnosis and link follow-up issues from #782.
|
||||
|
||||
## Results handoff
|
||||
|
||||
Pending: coordinator to attach the final SHA, artifact hashes, TRX counts/skips,
|
||||
interoperability/performance/runtime evidence, reviews and per-environment operator
|
||||
records in #782. This document makes no claim of a passing build, test, ScadaBridge,
|
||||
Windows/Linux smoke, live backend, soak or deployment. Required inaccessible steps
|
||||
remain open gates unless formally separated into linked issues with accurate scope.
|
||||
@@ -36,6 +36,37 @@ Transport security and OPC UA authentication are per-node concerns configured in
|
||||
|
||||
## Transport Security
|
||||
|
||||
### SDK 2.0.1 compatibility (#782)
|
||||
|
||||
The application targets exact OPC Foundation SDK **2.0.1**, replacing **1.5.378.182**.
|
||||
The upgrade retains the existing host, security profiles, anonymous-admission setting,
|
||||
API-key authentication and ACL flag semantics. It adds no companion-specification or
|
||||
PubSub capability. Release evidence and operator gates are linked from the
|
||||
[rollout runbook](plans/2026-11-10-opcua-sdk-2-rollout.md); source changes alone do not establish release approval.
|
||||
|
||||
SDK role augmentation preserves the application's principal, API-key id and application
|
||||
roles. A rejected API key must remain rejected rather than falling through to another
|
||||
identity provider. Per-IP authentication miss accounting uses the public transport-observed
|
||||
`SecureChannelContext.PeerAddress` during activation; an unknown peer retains the existing
|
||||
process-wide budget fallback. This is accounting evidence, not a client-supplied address or
|
||||
a new IP admission policy.
|
||||
|
||||
SDK event receive checks compose with application ACL filtering. For the application's
|
||||
rejected-write `AuditWriteUpdateEvent` only, a temporary validation context supplies
|
||||
`ReceiveEvents` to the recipient's SDK roles after the current source variable, effective
|
||||
identity, source `Read` grant and audit/encryption guard are checked. It expires before
|
||||
queueing; shared role-permission metadata is not widened. Prior permission callback
|
||||
errors veto delivery. SDK-generated audits retain SDK defaults. This does not enable
|
||||
auditing or grant a general audit feed.
|
||||
|
||||
Certificate configuration uses the SDK certificate manager, and certificate ownership
|
||||
lasts through the owning host/session's lifetime. Preserve the existing application
|
||||
certificate, private key, issuer/peer stores and filesystem permissions during rollout;
|
||||
verify key access and CA usage under the service identity. Do not rotate certificates,
|
||||
enable auto-accept, add `None`, or relax revocation checks to get the candidate running.
|
||||
Continuation points remain session-owned and bound to node/topology and current authority;
|
||||
restart requires fresh sessions and fresh points.
|
||||
|
||||
### Overview
|
||||
|
||||
The OtOpcUa Server supports configurable OPC UA transport security profiles that control how data is protected on the wire between OPC UA clients and the server.
|
||||
|
||||
+3
-3
@@ -114,7 +114,7 @@ public sealed class GatewayBoundingValueTests
|
||||
{
|
||||
var bound = await HistoryBounds.StartBoundAsync(Source(Aveva()), "T", At(-30), reverse: false, 65536, Now, Ct);
|
||||
|
||||
bound!.StatusCode.ShouldBe(Opc.Ua.StatusCodes.BadBoundNotFound);
|
||||
bound!.StatusCode.ShouldBe(Opc.Ua.StatusCodes.BadBoundNotFound.Code);
|
||||
bound.SourceTimestampUtc.ShouldBe(At(-30));
|
||||
bound.Value.ShouldBeNull();
|
||||
}
|
||||
@@ -134,7 +134,7 @@ public sealed class GatewayBoundingValueTests
|
||||
after.Value.ShouldBe(5d);
|
||||
|
||||
var missing = await HistoryBounds.EndBoundAsync(source, "T", HistoryBounds.For(At(0), At(101)), At(100), 65536, Now, Ct);
|
||||
missing.StatusCode.ShouldBe(Opc.Ua.StatusCodes.BadBoundNotFound);
|
||||
missing.StatusCode.ShouldBe(Opc.Ua.StatusCodes.BadBoundNotFound.Code);
|
||||
missing.SourceTimestampUtc.ShouldBe(At(101));
|
||||
}
|
||||
|
||||
@@ -156,7 +156,7 @@ public sealed class GatewayBoundingValueTests
|
||||
var calls = fake.ReadRawCallCount;
|
||||
var beyond = Now.AddDays(3);
|
||||
var missing = await HistoryBounds.EndBoundAsync(source, "T", HistoryBounds.For(At(0), beyond), At(100), 65536, Now, Ct);
|
||||
missing.StatusCode.ShouldBe(Opc.Ua.StatusCodes.BadBoundNotFound);
|
||||
missing.StatusCode.ShouldBe(Opc.Ua.StatusCodes.BadBoundNotFound.Code);
|
||||
missing.SourceTimestampUtc.ShouldBe(beyond);
|
||||
fake.ReadRawCallCount.ShouldBe(calls, "a bound beyond the horizon is not found without a read");
|
||||
source.GetHealthSnapshot().TotalFailures.ShouldBe(0);
|
||||
|
||||
+22
-19
@@ -106,7 +106,7 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
try
|
||||
{
|
||||
var first = ReadRaw(nm, nodeId, At(0), At(3600), 5000, null);
|
||||
first.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||||
first.Error.StatusCode.ShouldBe(StatusCodes.Good);
|
||||
first.Values.Count.ShouldBe(FirstPage);
|
||||
first.Cp.ShouldNotBeNull();
|
||||
// The budget the client asked for (+ lookahead + allowance, refused naming 1000), the ceiling re-read that cannot
|
||||
@@ -114,7 +114,7 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
fake.RawCalls.Select(c => c.MaxValues).ShouldBe([5000 + 1 + RawAggregateReader.PreWindowAllowance, Ceiling, Ceiling]);
|
||||
|
||||
var (rest, _, status) = DrainFrom(nm, nodeId, At(0), At(3600), 5000, first.Cp);
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
status.ShouldBe(StatusCodes.Good);
|
||||
first.Values.Concat(rest).ShouldBe(PerSecond(3600).Select(s => s.Item2));
|
||||
Refusals(fake).ShouldBe(1);
|
||||
}
|
||||
@@ -140,7 +140,7 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
? DrainFrom(nm, nodeId, At(3600), At(-1), n, null)
|
||||
: DrainFrom(nm, nodeId, At(0), At(3600), n, null);
|
||||
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
status.ShouldBe(StatusCodes.Good);
|
||||
var expected = stored.Select(s => s.Item2);
|
||||
values.ShouldBe(reverse ? expected.Reverse() : expected);
|
||||
pages[0].ShouldBe(FirstPage);
|
||||
@@ -217,10 +217,10 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
|
||||
var (values, pages, status) = DrainFrom(nm, nodeId, At(0), At(5000), n, null, maxPages: 60);
|
||||
var expected = stored.Select(s => s.Item2).ToList();
|
||||
if (status.Code == StatusCodes.Good) values.ShouldBe(expected);
|
||||
if (status == StatusCodes.Good) values.ShouldBe(expected);
|
||||
else
|
||||
{
|
||||
status.Code.ShouldBe(StatusCodes.BadTooManyOperations);
|
||||
status.ShouldBe(StatusCodes.BadTooManyOperations);
|
||||
values.ShouldBe(expected.Take(values.Count)); // an exact prefix
|
||||
size.ShouldBeGreaterThan(L - 2, "a cluster that fits the over-fetch is always served");
|
||||
}
|
||||
@@ -254,7 +254,7 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
ds.MaxRawValuesPerRead.ShouldBe(0);
|
||||
|
||||
var (rest, pages, status) = DrainFrom(nm, nodeId, At(0), At(5000), 0, first.Cp, maxPages: 20);
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
status.ShouldBe(StatusCodes.Good);
|
||||
first.Values.Concat(rest).ShouldBe(stored.Select(s => s.Item2));
|
||||
pages.Count.ShouldBeLessThan(20);
|
||||
}
|
||||
@@ -278,7 +278,7 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
fake.MaxValuesCeiling = 500; // the operator lowered it
|
||||
fake.RawCalls.Clear();
|
||||
var (rest, pages, status) = DrainFrom(nm, nodeId, At(0), At(3600), 5000, p2.Cp, maxPages: 20);
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
status.ShouldBe(StatusCodes.Good);
|
||||
got.AddRange(rest);
|
||||
got.ShouldBe(stored.Select(s => s.Item2));
|
||||
pages.ShouldAllBe(p => p <= 500 - RawAggregateReader.PreWindowAllowance);
|
||||
@@ -302,7 +302,7 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
var start = At(10).AddMilliseconds(500);
|
||||
var end = At(3000).AddMilliseconds(500);
|
||||
var (values, pages, status) = DrainFrom(nm, nodeId, start, end, 5000, null, returnBounds: true);
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
status.ShouldBe(StatusCodes.Good);
|
||||
values.ShouldBe(Enumerable.Range(10, 2992).Select(i => (double)i));
|
||||
pages.ShouldAllBe(p => p <= L);
|
||||
Refusals(fake).ShouldBe(0);
|
||||
@@ -325,7 +325,7 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
try
|
||||
{
|
||||
var r = ReadRaw(nm, nodeId, DateTime.MinValue, T0.AddMilliseconds(200_000), n, null);
|
||||
r.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||||
r.Error.StatusCode.ShouldBe(StatusCodes.Good);
|
||||
r.Cp.ShouldNotBeNull();
|
||||
var page = ceiling - RawAggregateReader.PreWindowAllowance - 1;
|
||||
r.Values.ShouldBe(Enumerable.Range(200_000 - r.Values.Count, r.Values.Count).Reverse().Select(i => (double)i));
|
||||
@@ -360,7 +360,7 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
try
|
||||
{
|
||||
var (values, pages, status) = DrainFrom(nm, nodeId, DateTime.UnixEpoch, DateTime.UnixEpoch.AddDays(1), 0, null);
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
status.ShouldBe(StatusCodes.Good);
|
||||
values.ShouldBe(stored.Select(s => s.Item2));
|
||||
pages.Count.ShouldBeGreaterThan(1);
|
||||
}
|
||||
@@ -377,7 +377,7 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
{
|
||||
ReadRaw(nm, nodeId, DateTime.UnixEpoch, DateTime.UnixEpoch.AddSeconds(5), 5000, null); // learns the ceiling
|
||||
var r = ReadRaw(nm, nodeId, DateTime.MinValue, DateTime.UnixEpoch.AddDays(30), 10, null);
|
||||
r.Error.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||||
r.Error.StatusCode.ShouldBe(StatusCodes.Good);
|
||||
r.Values.ShouldBe(Enumerable.Range(2991, 10).Reverse().Select(i => (double)i)); // the LATEST ten, newest first
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
@@ -425,7 +425,7 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
fake.RawCalls.Clear();
|
||||
|
||||
var (values, _, status) = DrainFrom(nm, nodeId, At(0), x.AddDays(1), n, null);
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
status.ShouldBe(StatusCodes.Good);
|
||||
values.ShouldBe(stored.Select(s => s.Item2));
|
||||
Refusals(fake).ShouldBe(0);
|
||||
EveryRequestFits(fake);
|
||||
@@ -476,7 +476,7 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
try
|
||||
{
|
||||
var r = ReadRaw(nm, nodeId, At(0), At(10), 10, null);
|
||||
r.Error.StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication);
|
||||
r.Error.StatusCode.ShouldBe(StatusCodes.BadNoCommunication);
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
@@ -528,7 +528,7 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
private static (IReadOnlyList<double> Values, ServiceResult Error, byte[]? Cp) ReadRaw(
|
||||
OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max, byte[]? inboundCp, bool returnBounds = false)
|
||||
{
|
||||
var context = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
|
||||
var context = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, requestLifetime: RequestLifetime.None, identity: null);
|
||||
var details = new ReadRawModifiedDetails
|
||||
{
|
||||
ReturnBounds = returnBounds,
|
||||
@@ -537,16 +537,19 @@ public sealed class GatewayCeilingPagingTests : IDisposable
|
||||
EndTime = end,
|
||||
NumValuesPerNode = max,
|
||||
};
|
||||
var nodesToRead = new List<HistoryReadValueId> { new() { NodeId = nodeId, ContinuationPoint = inboundCp } };
|
||||
var nodesToRead = new List<HistoryReadValueId> { new() { NodeId = nodeId, ContinuationPoint = inboundCp is null ? default : ByteString.From(inboundCp) } };
|
||||
var results = new List<SdkHistoryReadResult> { null! };
|
||||
var errors = new List<ServiceResult> { null! };
|
||||
nm.HistoryRead(context, details, TimestampsToReturn.Both, releaseContinuationPoints: false, nodesToRead, results, errors);
|
||||
|
||||
var values = new List<double>();
|
||||
if (results[0]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho))
|
||||
foreach (var dv in ((HistoryData)ExtensionObject.ToEncodeable(ho)).DataValues)
|
||||
values.Add(Convert.ToDouble(dv.Value));
|
||||
return (values, errors[0], results[0]?.ContinuationPoint);
|
||||
if (results[0]?.HistoryData is { IsNull: false } historyData)
|
||||
{
|
||||
var data = (HistoryData)ExtensionObject.ToEncodeable(historyData)!;
|
||||
foreach (var dv in data.DataValues)
|
||||
values.Add(Convert.ToDouble(SdkValueBoundary.ToClr(dv.WrappedValue)));
|
||||
}
|
||||
return (values, errors[0], results[0]?.ContinuationPoint is { IsNull: false } cp ? cp.ToArray() : null);
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
|
||||
+11
-8
@@ -205,7 +205,7 @@ public sealed class GatewayRawCapTests : IDisposable
|
||||
? DrainFrom(nm, nodeId, At(3600), At(-1), n, null)
|
||||
: DrainFrom(nm, nodeId, At(0), At(3600), n, null);
|
||||
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
status.ShouldBe(StatusCodes.Good);
|
||||
var expected = Enumerable.Range(0, 3600).Select(i => (double)i);
|
||||
values.ShouldBe(reverse ? expected.Reverse() : expected);
|
||||
pages.Count.ShouldBeGreaterThan(1, "a year-sized window at N = 0 is paged, not one shot");
|
||||
@@ -224,7 +224,7 @@ public sealed class GatewayRawCapTests : IDisposable
|
||||
try
|
||||
{
|
||||
var (values, pages, status) = DrainFrom(nm, nodeId, At(10).AddMilliseconds(500), At(3000).AddMilliseconds(500), 0, null, returnBounds: true);
|
||||
status.Code.ShouldBe(StatusCodes.Good);
|
||||
status.ShouldBe(StatusCodes.Good);
|
||||
values.ShouldBe(Enumerable.Range(10, 2992).Select(i => (double)i)); // start bound +10, data +11..+3000, end bound +3001
|
||||
pages.ShouldAllBe(p => p <= Cap);
|
||||
fake.RawCalls.ShouldAllBe(c => c.MaxValues > 0 && c.MaxValues <= Cap + RawAggregateReader.PreWindowAllowance);
|
||||
@@ -277,21 +277,24 @@ public sealed class GatewayRawCapTests : IDisposable
|
||||
private static (IReadOnlyList<double> Values, ServiceResult Error, byte[]? Cp) ReadRaw(
|
||||
OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max, byte[]? inboundCp, bool returnBounds = false)
|
||||
{
|
||||
var context = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
|
||||
var context = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, requestLifetime: RequestLifetime.None, identity: null);
|
||||
var details = new ReadRawModifiedDetails
|
||||
{
|
||||
ReturnBounds = returnBounds, IsReadModified = false, StartTime = start, EndTime = end, NumValuesPerNode = max,
|
||||
};
|
||||
var nodesToRead = new List<HistoryReadValueId> { new() { NodeId = nodeId, ContinuationPoint = inboundCp } };
|
||||
var nodesToRead = new List<HistoryReadValueId> { new() { NodeId = nodeId, ContinuationPoint = inboundCp is null ? default : ByteString.From(inboundCp) } };
|
||||
var results = new List<SdkHistoryReadResult> { null! };
|
||||
var errors = new List<ServiceResult> { null! };
|
||||
nm.HistoryRead(context, details, TimestampsToReturn.Both, releaseContinuationPoints: false, nodesToRead, results, errors);
|
||||
|
||||
var values = new List<double>();
|
||||
if (results[0]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho))
|
||||
foreach (var dv in ((HistoryData)ExtensionObject.ToEncodeable(ho)).DataValues)
|
||||
values.Add(Convert.ToDouble(dv.Value));
|
||||
return (values, errors[0], results[0]?.ContinuationPoint);
|
||||
if (results[0]?.HistoryData is { IsNull: false } historyData)
|
||||
{
|
||||
var data = (HistoryData)ExtensionObject.ToEncodeable(historyData)!;
|
||||
foreach (var dv in data.DataValues)
|
||||
values.Add(Convert.ToDouble(SdkValueBoundary.ToClr(dv.WrappedValue)));
|
||||
}
|
||||
return (values, errors[0], results[0]?.ContinuationPoint is { IsNull: false } cp ? cp.ToArray() : null);
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
|
||||
+156
@@ -0,0 +1,156 @@
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Client;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
|
||||
using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig;
|
||||
using CoreReadResult = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>Uses activated sessions and the production continuation store; denied resumes never reach a backend.</summary>
|
||||
public sealed class Sdk2HistoryAuthorityWireTests
|
||||
{
|
||||
private const string ValuePath = RawDeviceFolder + "/History";
|
||||
private static readonly DateTime Epoch = new(2026, 10, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
public static TheoryData<string, string> Cases => new(
|
||||
from kind in new[] { "raw", "modified", "events" }
|
||||
from change in new[] { "alias", "replace", "rebuild", "grant", "key", "expiry" }
|
||||
select (kind, change));
|
||||
|
||||
[Theory]
|
||||
[MemberData(nameof(Cases))]
|
||||
public async Task A_saved_cursor_requires_its_live_node_and_current_authority(string kind, string change)
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await StartAsync("HistoryAuthority", encrypted: true, ct);
|
||||
var source = new CountingSource();
|
||||
var clock = new Clock();
|
||||
var key = Key();
|
||||
key.ExpiresUtc = clock.GetUtcNow().UtcDateTime.AddMinutes(10);
|
||||
var grant = ClusterGrant(KeyPrincipal, "history-allow", NodePermissions.ReadOnly);
|
||||
var state = rig.Enforce(true, [key], grant);
|
||||
rig.Nm.Authorizer = new DataPlaneAuthorizer(state, new TriePermissionEvaluator(state.Tries), true,
|
||||
NullLogger.Instance, clock);
|
||||
rig.Nm.HistorianDataSource = source;
|
||||
Materialise(rig);
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct, encrypted: true);
|
||||
var first = await PageAsync(session, kind, alias: false, default, ct);
|
||||
first.StatusCode.ShouldBe(StatusCodes.Good);
|
||||
first.ContinuationPoint.IsEmpty.ShouldBeFalse("the production session store must own a real next page");
|
||||
source.Calls.ShouldBeGreaterThan(0);
|
||||
|
||||
switch (change)
|
||||
{
|
||||
case "replace":
|
||||
if (kind == "events") rig.Nm.RemoveEquipmentSubtree(RawDeviceFolder, AddressSpaceRealm.Raw).ShouldBeTrue();
|
||||
else rig.Nm.RemoveVariableNode(ValuePath, AddressSpaceRealm.Raw).ShouldBeTrue();
|
||||
Materialise(rig);
|
||||
break;
|
||||
case "rebuild":
|
||||
rig.Nm.RebuildAddressSpace(); Materialise(rig); rig.Nm.CompleteAddressSpaceApply();
|
||||
break;
|
||||
case "grant": Apply(state, key, []); break;
|
||||
case "key":
|
||||
var disabled = Key(); disabled.Enabled = false; Apply(state, disabled, [grant]); break;
|
||||
case "expiry": clock.Advance(TimeSpan.FromMinutes(20)); break;
|
||||
}
|
||||
var before = source.Calls;
|
||||
var denied = await PageAsync(session, kind, change == "alias", first.ContinuationPoint, ct);
|
||||
denied.StatusCode.ShouldBe(change is "grant" or "key" or "expiry"
|
||||
? StatusCodes.BadUserAccessDenied : StatusCodes.BadContinuationPointInvalid);
|
||||
denied.HistoryData.IsNull.ShouldBeTrue("a denied resume must not disclose saved data");
|
||||
denied.ContinuationPoint.IsEmpty.ShouldBeTrue();
|
||||
source.Calls.ShouldBe(before, "permission and exact-node binding must be checked before backend dispatch");
|
||||
|
||||
// A fresh authorized request remains usable: this is not an always-deny fixture.
|
||||
clock.Reset(); Apply(state, key, [grant]);
|
||||
(await PageAsync(session, kind, alias: false, default, ct)).StatusCode.ShouldBe(StatusCodes.Good);
|
||||
source.Calls.ShouldBeGreaterThan(before);
|
||||
await session.CloseAsync(ct);
|
||||
}
|
||||
|
||||
private static void Apply(DataPlaneAuthorityState state, ApiKey key, NodeAcl[] grants) =>
|
||||
state.Apply(ServerEventRig.Cluster, Guid.NewGuid().ToString("N"), [key], grants,
|
||||
new Dictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>());
|
||||
|
||||
private static void Materialise(ServerEventRig rig)
|
||||
{
|
||||
rig.MaterialiseNative();
|
||||
rig.Sink.EnsureVariable(ValuePath, RawDeviceFolder, "History", "Double", false, AddressSpaceRealm.Raw, "same-backend-tag");
|
||||
rig.Sink.EnsureVariable(ValuePath, EqA, "HistoryAlias", "Double", false, AddressSpaceRealm.Uns, "same-backend-tag");
|
||||
}
|
||||
|
||||
private static async Task<Opc.Ua.HistoryReadResult> PageAsync(ISession session, string kind, bool alias,
|
||||
ByteString point, CancellationToken ct)
|
||||
{
|
||||
var realm = alias ? V3NodeIds.UnsNamespaceUri : V3NodeIds.RawNamespaceUri;
|
||||
var path = kind == "events" ? alias ? EqA : RawDeviceFolder : ValuePath;
|
||||
var node = new NodeId(path, (ushort)session.NamespaceUris.GetIndex(realm));
|
||||
HistoryReadDetails details = kind == "events" ? new ReadEventDetails
|
||||
{
|
||||
StartTime = Epoch, EndTime = Epoch.AddMinutes(1), NumValuesPerNode = 2,
|
||||
Filter = new EventFilter { SelectClauses =
|
||||
[new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, QualifiedName.From(BrowseNames.EventId))] },
|
||||
} : new ReadRawModifiedDetails
|
||||
{ StartTime = Epoch, EndTime = Epoch.AddMinutes(1), NumValuesPerNode = 2, IsReadModified = kind == "modified" };
|
||||
var response = await session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, false,
|
||||
[new HistoryReadValueId { NodeId = node, ContinuationPoint = point }], ct);
|
||||
return response.Results.ToList().ShouldHaveSingleItem();
|
||||
}
|
||||
|
||||
private sealed class Clock : TimeProvider
|
||||
{
|
||||
private readonly DateTimeOffset _initial = DateTimeOffset.UtcNow;
|
||||
private TimeSpan _offset;
|
||||
public override DateTimeOffset GetUtcNow() => _initial + _offset;
|
||||
public void Advance(TimeSpan amount) => _offset += amount;
|
||||
public void Reset() => _offset = TimeSpan.Zero;
|
||||
}
|
||||
|
||||
private sealed class CountingSource : IHistorianDataSource
|
||||
{
|
||||
private int _calls;
|
||||
public int Calls => Volatile.Read(ref _calls);
|
||||
private static DataValueSnapshot Value(int i) => new((double)i, OpcUaStatus.Good, Epoch.AddSeconds(i), Epoch.AddSeconds(i));
|
||||
public Task<CoreReadResult> ReadRawAsync(string reference, DateTime start, DateTime end, uint max, CancellationToken ct)
|
||||
{
|
||||
Interlocked.Increment(ref _calls);
|
||||
var rows = Enumerable.Range(0, 25).Select(Value).Where(x => x.SourceTimestampUtc >= start && x.SourceTimestampUtc < end).ToList();
|
||||
var count = max == 0 ? rows.Count : checked((int)max);
|
||||
return Task.FromResult(new CoreReadResult(rows.Take(count).ToList(), rows.Count > count ? [1] : null));
|
||||
}
|
||||
public Task<HistoryModifiedReadResult> ReadModifiedAsync(string reference, DateTime start, DateTime end,
|
||||
uint max, HistoryModifiedCursor? resume, CancellationToken ct)
|
||||
{
|
||||
Interlocked.Increment(ref _calls);
|
||||
var rows = Enumerable.Range(0, 25).Where(i => Epoch.AddSeconds(i) >= start && Epoch.AddSeconds(i) < end
|
||||
&& (resume is null || i > resume.ModifiedId))
|
||||
.Select(i => new HistoryModifiedValue(Value(i), Epoch.AddMinutes(2), HistoryModificationType.Replace,
|
||||
"fixture", i, Epoch.AddSeconds(i))).ToList();
|
||||
var count = max == 0 ? rows.Count : checked((int)max);
|
||||
return Task.FromResult(new HistoryModifiedReadResult(rows.Take(count).ToList(), rows.Count > count));
|
||||
}
|
||||
public Task<HistoricalEventsResult> ReadEventsAsync(string? source, DateTime start, DateTime end, int max, CancellationToken ct)
|
||||
{
|
||||
Interlocked.Increment(ref _calls);
|
||||
var rows = Enumerable.Range(0, 25).Select(i => new HistoricalEvent($"e{i:00}", "src", Epoch.AddSeconds(i), Epoch, "m", 500))
|
||||
.Where(x => x.EventTimeUtc >= start && x.EventTimeUtc < end).ToList();
|
||||
var count = max <= 0 ? rows.Count : max;
|
||||
return Task.FromResult(new HistoricalEventsResult(rows.Take(count).ToList(), rows.Count > count ? [1] : null));
|
||||
}
|
||||
public Task<CoreReadResult> ReadProcessedAsync(string r, DateTime s, DateTime e, TimeSpan interval, HistoryAggregateType a, CancellationToken ct) =>
|
||||
NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, interval, a, ct);
|
||||
public Task<CoreReadResult> ReadAtTimeAsync(string r, IReadOnlyList<DateTime> times, CancellationToken ct) =>
|
||||
NullHistorianDataSource.Instance.ReadAtTimeAsync(r, times, ct);
|
||||
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
|
||||
public void Dispose() { }
|
||||
}
|
||||
}
|
||||
+542
@@ -0,0 +1,542 @@
|
||||
using System.Collections;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Diagnostics;
|
||||
using System.Globalization;
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Reflection;
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Client;
|
||||
using Opc.Ua.Configuration;
|
||||
using Opc.Ua.Server;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
|
||||
using ClientSession = Opc.Ua.Client.ISession;
|
||||
using Subscription = Opc.Ua.Client.Subscription;
|
||||
using MonitoredItem = Opc.Ua.Client.MonitoredItem;
|
||||
using ServerSession = Opc.Ua.Server.ISession;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// Opt-in Task 12 lifecycle evidence. Diagnostic mode is deliberately separate from the two-hour gate.
|
||||
/// Run only this test in an otherwise idle test process; RSS/FD counts describe the whole process.
|
||||
/// No key, certificate, store path, token, or OPC UA payload is written to the JSONL evidence.
|
||||
/// </summary>
|
||||
[Trait("Category", "E2E")]
|
||||
public sealed class Sdk2LifecycleSoakTests
|
||||
{
|
||||
private const string Scalar = ServerEventRig.RawDeviceFolder + "/SoakScalar";
|
||||
private const string ArrayTag = ServerEventRig.RawDeviceFolder + "/SoakArray";
|
||||
private const string UnsScalar = "SoakEquipment/Scalar";
|
||||
private static readonly DateTime Epoch = new(2026, 10, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
public static bool Enabled => Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK") == "1";
|
||||
|
||||
[Fact(SkipUnless = nameof(Enabled), Skip = "Opt in with OTOPCUA_SDK2_SOAK=1 and an explicit diagnostic/full mode.")]
|
||||
public async Task Repeated_lifecycles_return_resources_to_a_steady_baseline()
|
||||
{
|
||||
var mode = Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK_MODE");
|
||||
(mode is "diagnostic" or "full").ShouldBeTrue("explicit diagnostic or full mode is required");
|
||||
var seconds = int.Parse(Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK_SECONDS")
|
||||
?? (mode == "full" ? "7200" : "90"), CultureInfo.InvariantCulture);
|
||||
if (mode == "full") seconds.ShouldBeGreaterThanOrEqualTo(7200);
|
||||
else seconds.ShouldBeInRange(60, 120);
|
||||
var output = Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK_OUTPUT")
|
||||
?? throw new InvalidOperationException("An absolute, fresh OTOPCUA_SDK2_SOAK_OUTPUT JSONL path is required.");
|
||||
Path.IsPathFullyQualified(output).ShouldBeTrue();
|
||||
Directory.CreateDirectory(Path.GetDirectoryName(output)!);
|
||||
await using var evidence = new StreamWriter(new FileStream(output, FileMode.CreateNew, FileAccess.Write, FileShare.Read));
|
||||
using var run = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
|
||||
run.CancelAfter(TimeSpan.FromSeconds(seconds + 120));
|
||||
var ct = run.Token;
|
||||
var watch = Stopwatch.StartNew();
|
||||
var clientRoot = Path.Combine(Path.GetTempPath(), $"otua-soak-client-{Guid.NewGuid():N}");
|
||||
ApplicationInstance? application = null;
|
||||
ServerEventRig? rig = null;
|
||||
string? serverRoot = null;
|
||||
var cycles = 0;
|
||||
var restarts = 0;
|
||||
Sample? baseline = null;
|
||||
var retained = new Queue<Sample>();
|
||||
try
|
||||
{
|
||||
var telemetry = DefaultTelemetry.Create(_ => { });
|
||||
var security = TestClientSecurity.Build(clientRoot);
|
||||
security.ApplicationCertificate!.SubjectName = "CN=SDK2LifecycleSoak";
|
||||
var config = new ApplicationConfiguration(telemetry)
|
||||
{
|
||||
ApplicationName = "SDK2LifecycleSoak", ApplicationUri = $"urn:SDK2LifecycleSoak:{Guid.NewGuid():N}",
|
||||
ApplicationType = ApplicationType.Client, SecurityConfiguration = security,
|
||||
ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 15_000 },
|
||||
TransportQuotas = new TransportQuotas { OperationTimeout = 5000 },
|
||||
};
|
||||
application = new ApplicationInstance(telemetry) { ApplicationConfiguration = config };
|
||||
await config.ValidateAsync(ApplicationType.Client, ct);
|
||||
await application.CheckApplicationInstanceCertificatesAsync(true, null, ct);
|
||||
rig = await ServerEventRig.StartAsync("SDK2LifecycleSoak", encrypted: true, ct);
|
||||
serverRoot = RequiredField<string>(rig, "_pkiRoot");
|
||||
Configure(rig);
|
||||
// The required observation period excludes initial certificate creation and server startup.
|
||||
watch.Restart();
|
||||
await LogAsync(evidence, new
|
||||
{
|
||||
kind = "start", utc = DateTimeOffset.UtcNow, processId = Environment.ProcessId,
|
||||
mode, requestedSeconds = seconds, minimumFullSeconds = 7200,
|
||||
build = Environment.GetEnvironmentVariable("OTOPCUA_SOAK_BUILD_SHA"),
|
||||
sdk = typeof(OperationContext).Assembly.GetCustomAttribute<AssemblyInformationalVersionAttribute>()?.InformationalVersion,
|
||||
diagnosticInstrumentation = "SDK2.0.1 CP lists and certificate-validation cached stores; read-only",
|
||||
limits = new { retainedManagedGrowthBytes = 64L << 20, rssGrowthBytes = 128L << 20, fdGrowth = 32 },
|
||||
coverage = "encrypted sessions; scalar/Int16[] Raw+UNS; subscription removal; browse abandonment; history release/cross-session/close; surgical changes/rebuild; revocation; paused TCP read/recovery; same-PKI restart",
|
||||
});
|
||||
while (watch.Elapsed < TimeSpan.FromSeconds(seconds))
|
||||
{
|
||||
var cycle = Stopwatch.StartNew();
|
||||
await ExerciseAsync(rig, config, ++cycles, evidence, ct);
|
||||
await EventuallyAsync(() => PublicCounts(rig) == (0, 0, 0), ct);
|
||||
var sample = Capture(rig, config.CertificateManager, watch.Elapsed, cycles, clientRoot, serverRoot);
|
||||
sample.Sessions.ShouldBe(0); sample.Subscriptions.ShouldBe(0); sample.Items.ShouldBe(0);
|
||||
sample.BrowsePoints.ShouldBe(0); sample.HistoryPoints.ShouldBe(0);
|
||||
sample.LiveDataItems.ShouldBe(0); sample.LiveEventItems.ShouldBe(0); sample.MonitoredNodes.ShouldBe(0);
|
||||
sample.PkiDescriptors.ShouldBe(0, "quiescent certificate file descriptors must be closed");
|
||||
if (cycles >= 3)
|
||||
{
|
||||
baseline ??= sample;
|
||||
sample.CachedCertificateStores.ShouldBe(baseline.CachedCertificateStores);
|
||||
sample.Variables.ShouldBe(baseline.Variables); sample.Folders.ShouldBe(baseline.Folders);
|
||||
sample.Conditions.ShouldBe(baseline.Conditions);
|
||||
sample.Descriptors.ShouldBeLessThanOrEqualTo(baseline.Descriptors + 32);
|
||||
sample.ManagedBytes.ShouldBeLessThanOrEqualTo(baseline.ManagedBytes + (64L << 20));
|
||||
sample.RssBytes.ShouldBeLessThanOrEqualTo(baseline.RssBytes + (128L << 20));
|
||||
retained.Enqueue(sample);
|
||||
if (retained.Count > 24) retained.Dequeue();
|
||||
if (retained.Count == 24)
|
||||
{
|
||||
var window = retained.ToArray();
|
||||
var strictlyIncreasing = window.Zip(window.Skip(1)).All(pair => pair.Second.ManagedBytes > pair.First.ManagedBytes);
|
||||
(strictlyIncreasing && window[^1].ManagedBytes - window[0].ManagedBytes > (16L << 20)).ShouldBeFalse(
|
||||
"retained managed memory must not grow monotonically across 24 quiescent cycles");
|
||||
}
|
||||
}
|
||||
await LogAsync(evidence, new { kind = "quiescent", cycleSeconds = cycle.Elapsed.TotalSeconds, resources = sample });
|
||||
if (cycles % 4 == 0)
|
||||
{
|
||||
var restart = Stopwatch.StartNew();
|
||||
await RestartWithActiveResourcesAsync(rig, config, ct).WaitAsync(TimeSpan.FromSeconds(25), ct);
|
||||
Configure(rig);
|
||||
restarts++;
|
||||
await LogAsync(evidence, new { kind = "restart", cycles, restarts, seconds = restart.Elapsed.TotalSeconds });
|
||||
}
|
||||
}
|
||||
cycles.ShouldBeGreaterThanOrEqualTo(4, "diagnostic mode must exercise at least one actual restart");
|
||||
restarts.ShouldBeGreaterThan(0);
|
||||
watch.Elapsed.TotalSeconds.ShouldBeGreaterThanOrEqualTo(seconds);
|
||||
var finalClientCores = CertificateCores(config.CertificateManager);
|
||||
var finalServerCores = CertificateCores(rig.Server.CertificateManager!);
|
||||
await rig.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), ct);
|
||||
rig = null;
|
||||
await application.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), ct);
|
||||
application = null;
|
||||
finalClientCores.Sum(CachedStores).ShouldBe(0);
|
||||
finalServerCores.Sum(CachedStores).ShouldBe(0);
|
||||
DescriptorCounts(clientRoot, serverRoot).Pki.ShouldBe(0);
|
||||
await LogAsync(evidence, new { kind = "complete", utc = DateTimeOffset.UtcNow, mode, cycles, restarts, seconds = watch.Elapsed.TotalSeconds,
|
||||
satisfiesTwoHourDuration = mode == "full" && watch.Elapsed.TotalSeconds >= 7200 });
|
||||
}
|
||||
catch (Exception error)
|
||||
{
|
||||
await LogAsync(evidence, new { kind = "failed", cycles, restarts, seconds = watch.Elapsed.TotalSeconds,
|
||||
exceptionType = error.GetType().FullName });
|
||||
throw;
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (rig is not null) await rig.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), TestContext.Current.CancellationToken);
|
||||
if (application is not null) await application.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), TestContext.Current.CancellationToken);
|
||||
if (Directory.Exists(clientRoot)) Directory.Delete(clientRoot, true);
|
||||
if (serverRoot is not null) Directory.Exists(serverRoot).ShouldBeFalse("the task-owned server PKI must be released");
|
||||
}
|
||||
}
|
||||
|
||||
private static void Configure(ServerEventRig rig)
|
||||
{
|
||||
rig.MaterialiseNative();
|
||||
rig.Nm.HistorianDataSource = new EventHistoryPagingOverTheWireTests.SqlLikeEventSource(Enumerable.Range(0, 40)
|
||||
.Select(i => new HistoricalEvent($"soak-{i}", "source", Epoch.AddSeconds(i), Epoch, "soak", 500)).ToList());
|
||||
rig.Nm.NodeWriteGateway = new AcceptWrites();
|
||||
MaterialiseValues(rig);
|
||||
rig.Enforce(true, [ServerEventRig.Key()], Grant());
|
||||
}
|
||||
|
||||
private static ZB.MOM.WW.OtOpcUa.Configuration.Entities.NodeAcl Grant() => ServerEventRig.ClusterGrant(
|
||||
ServerEventRig.KeyPrincipal, "soak-grant", NodePermissions.Browse | NodePermissions.Read | NodePermissions.Subscribe
|
||||
| NodePermissions.WriteOperate | NodePermissions.HistoryRead);
|
||||
|
||||
private static void MaterialiseValues(ServerEventRig rig)
|
||||
{
|
||||
rig.Sink.EnsureFolder(ServerEventRig.RawDeviceFolder, null, "Device", AddressSpaceRealm.Raw);
|
||||
rig.Sink.EnsureVariable(Scalar, ServerEventRig.RawDeviceFolder, "Scalar", "Int32", true, AddressSpaceRealm.Raw);
|
||||
rig.Sink.EnsureVariable(ArrayTag, ServerEventRig.RawDeviceFolder, "Array", "Int16", true, AddressSpaceRealm.Raw,
|
||||
isArray: true, arrayLength: 4);
|
||||
rig.Sink.EnsureFolder("SoakEquipment", null, "Equipment", AddressSpaceRealm.Uns);
|
||||
rig.Sink.EnsureVariable(UnsScalar, "SoakEquipment", "Scalar", "Int32", true, AddressSpaceRealm.Uns);
|
||||
for (var i = 0; i < 16; i++) rig.Sink.EnsureVariable($"{ServerEventRig.RawDeviceFolder}/Browse{i}",
|
||||
ServerEventRig.RawDeviceFolder, $"Browse{i}", "Int32", false, AddressSpaceRealm.Raw);
|
||||
rig.Sink.WriteValue(Scalar, 0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
|
||||
rig.Sink.WriteValue(ArrayTag, new short[] { 0, 1, 2, 3 }, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
|
||||
rig.Sink.WriteValue(UnsScalar, 0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);
|
||||
}
|
||||
|
||||
private static async Task ExerciseAsync(ServerEventRig rig, ApplicationConfiguration config, int cycle,
|
||||
StreamWriter evidence, CancellationToken ct)
|
||||
{
|
||||
await using var proxy = new PauseProxy(new Uri(rig.EndpointUrl).Port);
|
||||
using var owner = await ConnectAsync(rig, config, proxy.Port, ct);
|
||||
using var other = await ConnectAsync(rig, config, null, ct);
|
||||
var ownedSessions = rig.Server.CurrentInstance.SessionManager.GetSessions().ToArray();
|
||||
ownedSessions.Length.ShouldBe(2);
|
||||
var scalar = Raw(owner, Scalar);
|
||||
var array = Raw(owner, ArrayTag);
|
||||
var unsNs = owner.NamespaceUris.GetIndex(V3NodeIds.UnsNamespaceUri);
|
||||
unsNs.ShouldBeGreaterThan(0);
|
||||
var uns = new NodeId(UnsScalar, (ushort)unsNs);
|
||||
owner.NamespaceUris.GetString(scalar.NamespaceIndex).ShouldBe(V3NodeIds.RawNamespaceUri);
|
||||
owner.NamespaceUris.GetString(uns.NamespaceIndex).ShouldBe(V3NodeIds.UnsNamespaceUri);
|
||||
using var subscription = new Subscription(owner.DefaultSubscription) { PublishingInterval = 100, KeepAliveCount = 10 };
|
||||
owner.AddSubscription(subscription);
|
||||
try
|
||||
{
|
||||
await subscription.CreateAsync(ct);
|
||||
var seen = new ConcurrentDictionary<NodeId, Variant>();
|
||||
foreach (var id in new[] { scalar, array, uns })
|
||||
{
|
||||
var item = new MonitoredItem(subscription.DefaultItem) { StartNodeId = id, AttributeId = Attributes.Value,
|
||||
SamplingInterval = 50, QueueSize = 10 };
|
||||
item.Notification += (sender, _) =>
|
||||
{
|
||||
foreach (var value in sender.DequeueValues()) if (StatusCode.IsGood(value.StatusCode)) seen[id] = value.WrappedValue;
|
||||
};
|
||||
subscription.AddItem(item);
|
||||
}
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
subscription.MonitoredItems.ShouldAllBe(item => ServiceResult.IsGood(item.Status.Error));
|
||||
PublicCounts(rig).ShouldBe((2, 1, 3));
|
||||
var values = new short[] { (short)(cycle % 30000), 2, 3, 4 };
|
||||
var writes = await owner.WriteAsync(null, new[]
|
||||
{
|
||||
new WriteValue { NodeId = scalar, AttributeId = Attributes.Value, Value = new DataValue(Variant.From(cycle)) },
|
||||
new WriteValue { NodeId = array, AttributeId = Attributes.Value, Value = new DataValue(SdkTestValues.FromClr(values)) },
|
||||
new WriteValue { NodeId = uns, AttributeId = Attributes.Value, Value = new DataValue(Variant.From(cycle)) },
|
||||
}, ct);
|
||||
writes.Results.ToList().ShouldAllBe(status => status == StatusCodes.Good);
|
||||
(await owner.ReadValueAsync(scalar, ct)).WrappedValue.GetInt32().ShouldBe(cycle);
|
||||
(await owner.ReadValueAsync(array, ct)).ClrValue().ShouldBeOfType<short[]>().ShouldBe(values);
|
||||
(await owner.ReadValueAsync(uns, ct)).WrappedValue.GetInt32().ShouldBe(cycle);
|
||||
await EventuallyAsync(() => seen.TryGetValue(scalar, out var value) && value.GetInt32() == cycle
|
||||
&& seen.ContainsKey(array) && seen.TryGetValue(uns, out var mirror) && mirror.GetInt32() == cycle, ct);
|
||||
|
||||
// One explicitly released and one deliberately abandoned browse point.
|
||||
var browse = await BrowseAsync(owner, ct);
|
||||
browse.IsEmpty.ShouldBeFalse();
|
||||
(await owner.BrowseNextAsync(null, true, [browse], ct)).Results[0].StatusCode.ShouldBe(StatusCodes.Good);
|
||||
(await BrowseAsync(owner, ct)).IsEmpty.ShouldBeFalse();
|
||||
// Production history points: cross-session denial, single-use resume, explicit release and close cleanup.
|
||||
var first = (await PageAsync(owner, default, false, ct)).ContinuationPoint;
|
||||
first.IsEmpty.ShouldBeFalse();
|
||||
(await PageAsync(other, first, false, ct)).StatusCode.ShouldBe(StatusCodes.BadContinuationPointInvalid);
|
||||
var next = (await PageAsync(owner, first, false, ct)).ContinuationPoint;
|
||||
next.IsEmpty.ShouldBeFalse();
|
||||
(await PageAsync(owner, next, true, ct)).StatusCode.ShouldBe(StatusCodes.Good);
|
||||
for (var i = 0; i < 6; i++) (await PageAsync(owner, default, false, ct)).ContinuationPoint.IsEmpty.ShouldBeFalse();
|
||||
var held = ownedSessions.Select(ContinuationCounts).ToArray();
|
||||
held.Sum(count => count.Browse).ShouldBeGreaterThan(0);
|
||||
held.Sum(count => count.History).ShouldBeGreaterThan(0);
|
||||
var active = PublicCounts(rig);
|
||||
await LogAsync(evidence, new { kind = "active", cycle, sessions = active.Sessions,
|
||||
subscriptions = active.Subscriptions, items = active.Items, browsePoints = held.Sum(x => x.Browse), historyPoints = held.Sum(x => x.History) });
|
||||
|
||||
rig.Nm.UpdateTagAttributes(Scalar, false, null, "Int32", false, null, AddressSpaceRealm.Raw).ShouldBeTrue();
|
||||
rig.Nm.UpdateTagAttributes(Scalar, true, null, "Int32", false, null, AddressSpaceRealm.Raw).ShouldBeTrue();
|
||||
rig.Nm.RemoveVariableNode(ArrayTag, AddressSpaceRealm.Raw).ShouldBeTrue();
|
||||
rig.Sink.EnsureVariable(ArrayTag, ServerEventRig.RawDeviceFolder, "Array", "Int16", true, AddressSpaceRealm.Raw,
|
||||
isArray: true, arrayLength: 4);
|
||||
rig.Sink.WriteValue(ArrayTag, values, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
|
||||
var authority = rig.Nm.DataPlaneAuthority.ShouldBeOfType<DataPlaneAuthorityState>();
|
||||
authority.Apply(ServerEventRig.Cluster, $"revoked-{cycle}", [], [Grant()],
|
||||
new Dictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>());
|
||||
var denied = await owner.ReadAsync(null, 0, TimestampsToReturn.Neither,
|
||||
[new ReadValueId { NodeId = scalar, AttributeId = Attributes.Value }], ct);
|
||||
denied.Results[0].StatusCode.ShouldBe(StatusCodes.BadUserAccessDenied);
|
||||
rig.Enforce(true, [ServerEventRig.Key()], Grant());
|
||||
|
||||
proxy.Pause();
|
||||
var stalled = Stopwatch.StartNew();
|
||||
string stalledStatus;
|
||||
try
|
||||
{
|
||||
using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||||
timeout.CancelAfter(TimeSpan.FromSeconds(1));
|
||||
await owner.ReadValueAsync(scalar, timeout.Token).WaitAsync(TimeSpan.FromSeconds(5), ct);
|
||||
throw new InvalidOperationException("The deliberately paused network unexpectedly completed the read.");
|
||||
}
|
||||
catch (OperationCanceledException) when (!ct.IsCancellationRequested) { stalledStatus = "OperationCanceled"; }
|
||||
// Exact-tag ChannelAsyncOperation maps a canceled completion waiter to BadRequestInterrupted.
|
||||
catch (ServiceResultException error) when (error.StatusCode == StatusCodes.BadTimeout
|
||||
|| error.StatusCode == StatusCodes.BadRequestCancelledByClient || error.StatusCode == StatusCodes.BadRequestTimeout
|
||||
|| error.StatusCode == StatusCodes.BadRequestInterrupted) { stalledStatus = error.StatusCode.ToString(); }
|
||||
finally { proxy.Resume(); }
|
||||
(await owner.ReadValueAsync(scalar, ct)).StatusCode.ShouldBe(StatusCodes.Good);
|
||||
await LogAsync(evidence, new { kind = "network-stall-recovered", cycle, status = stalledStatus, seconds = stalled.Elapsed.TotalSeconds });
|
||||
await subscription.DeleteAsync(true, ct);
|
||||
await owner.RemoveSubscriptionAsync(subscription, ct);
|
||||
rig.Sink.RebuildAddressSpace();
|
||||
Configure(rig);
|
||||
}
|
||||
finally
|
||||
{
|
||||
proxy.Resume();
|
||||
await owner.CloseAsync(ct).WaitAsync(TimeSpan.FromSeconds(10), ct);
|
||||
await other.CloseAsync(ct).WaitAsync(TimeSpan.FromSeconds(10), ct);
|
||||
foreach (var closed in ownedSessions)
|
||||
{
|
||||
ContinuationCounts(closed).ShouldBe((0, 0));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task RestartWithActiveResourcesAsync(ServerEventRig rig, ApplicationConfiguration config,
|
||||
CancellationToken ct)
|
||||
{
|
||||
using var client = await ConnectAsync(rig, config, null, ct);
|
||||
using var subscription = new Subscription(client.DefaultSubscription) { PublishingInterval = 100 };
|
||||
client.AddSubscription(subscription);
|
||||
await subscription.CreateAsync(ct);
|
||||
subscription.AddItem(new MonitoredItem(subscription.DefaultItem)
|
||||
{
|
||||
StartNodeId = Raw(client, Scalar), AttributeId = Attributes.Value, SamplingInterval = 50,
|
||||
});
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
PublicCounts(rig).ShouldBe((1, 1, 1));
|
||||
(await BrowseAsync(client, ct)).IsEmpty.ShouldBeFalse();
|
||||
(await PageAsync(client, default, false, ct)).ContinuationPoint.IsEmpty.ShouldBeFalse();
|
||||
var oldInternal = rig.Server.CurrentInstance;
|
||||
var oldSessionManager = oldInternal.SessionManager;
|
||||
var oldSubscriptionManager = oldInternal.SubscriptionManager;
|
||||
var oldSessions = oldSessionManager.GetSessions().ToArray();
|
||||
var oldSubscriptions = oldSubscriptionManager.GetSubscriptions().ToArray();
|
||||
var oldManager = rig.Server.CertificateManager!;
|
||||
var oldCores = CertificateCores(oldManager);
|
||||
await rig.RestartAsync(ct).WaitAsync(TimeSpan.FromSeconds(20), ct);
|
||||
oldSessionManager.GetSessions().ShouldBeEmpty();
|
||||
oldSubscriptionManager.GetSubscriptions().ShouldBeEmpty();
|
||||
oldSubscriptions.ShouldAllBe(item => item.MonitoredItemCount == 0);
|
||||
foreach (var session in oldSessions) ContinuationCounts(session).ShouldBe((0, 0));
|
||||
RequiredField<bool>(oldManager, "m_disposed").ShouldBeTrue();
|
||||
oldCores.Sum(CachedStores).ShouldBe(0, "stopped server validators must release every cached store");
|
||||
PublicCounts(rig).ShouldBe((0, 0, 0));
|
||||
}
|
||||
|
||||
private static async Task<ClientSession> ConnectAsync(ServerEventRig rig, ApplicationConfiguration config,
|
||||
int? proxyPort, CancellationToken ct)
|
||||
{
|
||||
var telemetry = DefaultTelemetry.Create(_ => { });
|
||||
var endpoint = await CoreClientUtils.SelectEndpointAsync(config, rig.EndpointUrl, true, telemetry, ct)
|
||||
?? throw new InvalidOperationException("No encrypted endpoint.");
|
||||
endpoint.SecurityMode.ShouldBe(MessageSecurityMode.SignAndEncrypt);
|
||||
if (proxyPort is { } port) endpoint.EndpointUrl = $"opc.tcp://127.0.0.1:{port}/OtOpcUa";
|
||||
return await new DefaultSessionFactory(telemetry).CreateAsync(config,
|
||||
new ConfiguredEndpoint(null, endpoint, EndpointConfiguration.Create(config)), false, false,
|
||||
"SDK2LifecycleSoak", 15_000, new UserIdentity("apikey", Encoding.UTF8.GetBytes(ServerEventRig.KeyPassword)), default, ct);
|
||||
}
|
||||
|
||||
private static NodeId Raw(ClientSession session, string id)
|
||||
{
|
||||
var ns = session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri);
|
||||
ns.ShouldBeGreaterThan(0);
|
||||
return new NodeId(id, (ushort)ns);
|
||||
}
|
||||
|
||||
private static async Task<ByteString> BrowseAsync(ClientSession session, CancellationToken ct) =>
|
||||
(await session.BrowseAsync(null, null, 1, [new BrowseDescription
|
||||
{
|
||||
NodeId = Raw(session, ServerEventRig.RawDeviceFolder), BrowseDirection = BrowseDirection.Forward,
|
||||
ReferenceTypeId = ReferenceTypeIds.HierarchicalReferences, IncludeSubtypes = true,
|
||||
ResultMask = (uint)BrowseResultMask.All,
|
||||
}], ct)).Results[0].ContinuationPoint;
|
||||
|
||||
private static async Task<Opc.Ua.HistoryReadResult> PageAsync(ClientSession session, ByteString point,
|
||||
bool release, CancellationToken ct) => (await session.HistoryReadAsync(null, new ExtensionObject(new ReadEventDetails
|
||||
{
|
||||
StartTime = Epoch.AddDays(-1), EndTime = Epoch.AddDays(1), NumValuesPerNode = 4,
|
||||
Filter = new EventFilter { SelectClauses = [new SimpleAttributeOperand(ObjectTypeIds.BaseEventType,
|
||||
QualifiedName.From(BrowseNames.EventId))] },
|
||||
}), TimestampsToReturn.Both, release,
|
||||
[new HistoryReadValueId { NodeId = Raw(session, ServerEventRig.RawDeviceFolder), ContinuationPoint = point }], ct)).Results[0];
|
||||
|
||||
private static (int Sessions, int Subscriptions, int Items) PublicCounts(ServerEventRig rig)
|
||||
{
|
||||
var subscriptions = rig.Server.CurrentInstance.SubscriptionManager.GetSubscriptions();
|
||||
return (rig.Server.CurrentInstance.SessionManager.GetSessions().Count, subscriptions.Count,
|
||||
subscriptions.Sum(subscription => subscription.MonitoredItemCount));
|
||||
}
|
||||
|
||||
// Exact-tag diagnostic instrumentation only. Fail if fields/types change; never replace or mutate a store.
|
||||
private static (int Browse, int History) ContinuationCounts(ServerSession session)
|
||||
{
|
||||
var points = session.ContinuationPoints;
|
||||
using (RequiredField<System.Threading.Lock>(points, "m_lock").EnterScope())
|
||||
return (NullableCollectionCount(points, "m_browse"), NullableCollectionCount(points, "m_history"));
|
||||
}
|
||||
|
||||
private static object[] CertificateCores(object manager)
|
||||
{
|
||||
using (RequiredField<System.Threading.Lock>(manager, "m_certificatesLock").EnterScope())
|
||||
{
|
||||
var cores = new List<object>();
|
||||
foreach (var field in new[] { "m_peerCore", "m_userCore", "m_httpsCore" })
|
||||
if (Field(manager, field).GetValue(manager) is { } core) cores.Add(core);
|
||||
cores.AddRange(RequiredField<IDictionary>(manager, "m_customCores").Values.Cast<object>());
|
||||
return cores.Distinct(System.Collections.Generic.ReferenceEqualityComparer.Instance).ToArray();
|
||||
}
|
||||
}
|
||||
|
||||
private static int CachedStores(object core) => RequiredField<IDictionary>(core, "m_stores").Count;
|
||||
private static int NullableCollectionCount(object owner, string name) => Field(owner, name).GetValue(owner) switch
|
||||
{
|
||||
null => 0,
|
||||
ICollection items => items.Count,
|
||||
_ => throw new InvalidOperationException($"SDK diagnostic field {name} no longer has the expected collection type."),
|
||||
};
|
||||
private static FieldInfo Field(object owner, string name) => owner.GetType().GetField(name,
|
||||
BindingFlags.Instance | BindingFlags.NonPublic) ?? throw new InvalidOperationException($"Missing SDK diagnostic field {name}.");
|
||||
private static T RequiredField<T>(object owner, string name) => Field(owner, name).GetValue(owner) is T value ? value
|
||||
: throw new InvalidOperationException($"SDK diagnostic field {name} no longer has the expected type {typeof(T).Name}.");
|
||||
|
||||
private sealed record Sample(int Cycle, double Seconds, int Sessions, int Subscriptions, int Items,
|
||||
int BrowsePoints, int HistoryPoints, int LiveDataItems, int LiveEventItems, int MonitoredNodes,
|
||||
int Variables, int Folders, int Conditions, int CachedCertificateStores,
|
||||
int Descriptors, int PkiDescriptors, long ManagedBytes, long RssBytes, long AllocatedBytes,
|
||||
int Gen0, int Gen1, int Gen2);
|
||||
|
||||
private static Sample Capture(ServerEventRig rig, object clientManager, TimeSpan elapsed, int cycle,
|
||||
string clientRoot, string serverRoot)
|
||||
{
|
||||
var counts = PublicCounts(rig);
|
||||
var points = rig.Server.CurrentInstance.SessionManager.GetSessions().Select(ContinuationCounts).ToArray();
|
||||
var descriptors = DescriptorCounts(clientRoot, serverRoot);
|
||||
var retained = GC.GetTotalMemory(forceFullCollection: true);
|
||||
using var process = Process.GetCurrentProcess();
|
||||
process.Refresh();
|
||||
return new Sample(cycle, elapsed.TotalSeconds, counts.Sessions, counts.Subscriptions, counts.Items,
|
||||
points.Sum(x => x.Browse), points.Sum(x => x.History), rig.Nm.LiveItemCount, rig.Nm.LiveEventItemCount,
|
||||
rig.Nm.MonitoredNodeCount, rig.Nm.VariableCount, rig.Nm.FolderCount, rig.Nm.AlarmConditionCount,
|
||||
CertificateCores(rig.Server.CertificateManager!).Sum(CachedStores) + CertificateCores(clientManager).Sum(CachedStores),
|
||||
descriptors.Total, descriptors.Pki, retained, process.WorkingSet64, GC.GetTotalAllocatedBytes(),
|
||||
GC.CollectionCount(0), GC.CollectionCount(1), GC.CollectionCount(2));
|
||||
}
|
||||
|
||||
private static (int Total, int Pki) DescriptorCounts(params string[] roots)
|
||||
{
|
||||
if (OperatingSystem.IsLinux())
|
||||
{
|
||||
var targets = Directory.EnumerateFileSystemEntries("/proc/self/fd").Select(path => new FileInfo(path).LinkTarget).ToArray();
|
||||
return (targets.Length, targets.Count(path => path is not null && roots.Any(root => path.StartsWith(root, StringComparison.Ordinal))));
|
||||
}
|
||||
if (OperatingSystem.IsMacOS())
|
||||
{
|
||||
using var process = new Process { StartInfo = new ProcessStartInfo("/usr/sbin/lsof")
|
||||
{ RedirectStandardOutput = true, RedirectStandardError = true, UseShellExecute = false } };
|
||||
foreach (var arg in new[] { "-a", "-p", Environment.ProcessId.ToString(CultureInfo.InvariantCulture), "-Fnf" })
|
||||
process.StartInfo.ArgumentList.Add(arg);
|
||||
process.Start();
|
||||
var text = process.StandardOutput.ReadToEnd();
|
||||
_ = process.StandardError.ReadToEnd();
|
||||
if (!process.WaitForExit(5000)) { process.Kill(); throw new TimeoutException("Descriptor probe stalled."); }
|
||||
process.ExitCode.ShouldBe(0);
|
||||
var lines = text.Split('\n');
|
||||
return (lines.Count(line => line.Length > 1 && line[0] == 'f' && char.IsDigit(line[1])),
|
||||
lines.Count(line => line.StartsWith('n') && roots.Any(root => line.AsSpan(1).StartsWith(root, StringComparison.Ordinal))));
|
||||
}
|
||||
throw new PlatformNotSupportedException("This soak requires Linux procfs or macOS lsof for actual descriptor evidence.");
|
||||
}
|
||||
|
||||
private static async Task EventuallyAsync(Func<bool> condition, CancellationToken ct)
|
||||
{
|
||||
var deadline = Stopwatch.StartNew();
|
||||
while (!condition() && deadline.Elapsed < TimeSpan.FromSeconds(10)) await Task.Delay(25, ct);
|
||||
condition().ShouldBeTrue("resource or notification condition did not settle within ten seconds");
|
||||
}
|
||||
private static async Task LogAsync(StreamWriter output, object value)
|
||||
{
|
||||
await output.WriteLineAsync(JsonSerializer.Serialize(value));
|
||||
await output.FlushAsync();
|
||||
}
|
||||
private sealed class AcceptWrites : IOpcUaNodeWriteGateway
|
||||
{
|
||||
public Task<NodeWriteOutcome> WriteAsync(string id, object? value, AddressSpaceRealm realm, CancellationToken ct) =>
|
||||
Task.FromResult(new NodeWriteOutcome(true, null));
|
||||
}
|
||||
|
||||
/// <summary>Task-owned TCP relay. Pause blocks forwarding in both directions without closing either socket.</summary>
|
||||
private sealed class PauseProxy : IAsyncDisposable
|
||||
{
|
||||
private readonly TcpListener _listener = new(IPAddress.Loopback, 0);
|
||||
private readonly CancellationTokenSource _stop = new();
|
||||
private readonly ConcurrentBag<TcpClient> _sockets = [];
|
||||
private readonly ConcurrentBag<Task> _pumps = [];
|
||||
private readonly Task _accept;
|
||||
private TaskCompletionSource? _pause;
|
||||
internal int Port => ((IPEndPoint)_listener.LocalEndpoint).Port;
|
||||
internal PauseProxy(int targetPort)
|
||||
{
|
||||
_listener.Start();
|
||||
_accept = AcceptAsync(targetPort);
|
||||
}
|
||||
internal void Pause() => Volatile.Write(ref _pause, new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously));
|
||||
internal void Resume() => Interlocked.Exchange(ref _pause, null)?.TrySetResult();
|
||||
private async Task AcceptAsync(int port)
|
||||
{
|
||||
try
|
||||
{
|
||||
while (!_stop.IsCancellationRequested)
|
||||
{
|
||||
var incoming = await _listener.AcceptTcpClientAsync(_stop.Token);
|
||||
_sockets.Add(incoming);
|
||||
var outgoing = new TcpClient(); _sockets.Add(outgoing);
|
||||
await outgoing.ConnectAsync(IPAddress.Loopback, port, _stop.Token);
|
||||
_pumps.Add(PumpAsync(incoming.GetStream(), outgoing.GetStream()));
|
||||
_pumps.Add(PumpAsync(outgoing.GetStream(), incoming.GetStream()));
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) when (_stop.IsCancellationRequested) { }
|
||||
}
|
||||
private async Task PumpAsync(NetworkStream source, NetworkStream destination)
|
||||
{
|
||||
var buffer = new byte[16 * 1024];
|
||||
try
|
||||
{
|
||||
while (true)
|
||||
{
|
||||
var length = await source.ReadAsync(buffer, _stop.Token);
|
||||
if (length == 0) return;
|
||||
if (Volatile.Read(ref _pause) is { } paused) await paused.Task.WaitAsync(_stop.Token);
|
||||
await destination.WriteAsync(buffer.AsMemory(0, length), _stop.Token);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) when (_stop.IsCancellationRequested) { }
|
||||
catch (IOException) when (_stop.IsCancellationRequested) { }
|
||||
}
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
Resume(); _stop.Cancel(); _listener.Stop();
|
||||
foreach (var socket in _sockets) socket.Dispose();
|
||||
try { await _accept; await Task.WhenAll(_pumps); }
|
||||
finally { _stop.Dispose(); }
|
||||
}
|
||||
}
|
||||
}
|
||||
+19
@@ -1,4 +1,6 @@
|
||||
using System.Text;
|
||||
using System.Net;
|
||||
using System.Collections.Concurrent;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Client;
|
||||
@@ -29,6 +31,16 @@ public sealed class Sdk2TrustedChannelTests
|
||||
(await session.ReadValueAsync(rig.Node(session), ct)).WrappedValue.GetInt32().ShouldBe(7);
|
||||
(await rig.WriteAsync(session, 42, ct)).ShouldBe(StatusCodes.Good);
|
||||
rig.Writes.ShouldBe(1, "a narrow grant with no application-wide roles must permit the write");
|
||||
var observedPeer = rig.Peers.ShouldHaveSingleItem();
|
||||
observedPeer.ShouldNotBeNull();
|
||||
IPAddress.IsLoopback(observedPeer.Address).ShouldBeTrue();
|
||||
observedPeer.Port.ShouldBe(0, "the public SDK evidence supplies an IP address, not an invented client port");
|
||||
var originalSessionId = session.SessionId;
|
||||
await session.ReconnectAsync(null, null, ct);
|
||||
session.SessionId.ShouldBe(originalSessionId, "the test must reactivate the existing session on a new channel");
|
||||
rig.Peers.Count.ShouldBe(2);
|
||||
rig.Peers.Last().ShouldNotBeNull().Address.ShouldBe(observedPeer.Address);
|
||||
(await session.ReadValueAsync(rig.Node(session), ct)).StatusCode.ShouldBe(StatusCodes.Good);
|
||||
var identity = rig.Server.CurrentInstance.SessionManager.GetSessions().ShouldHaveSingleItem().EffectiveIdentity
|
||||
.ShouldBeOfType<RoleCarryingUserIdentity>();
|
||||
identity.Principal.ShouldBe(TrustedRig.Principal);
|
||||
@@ -80,6 +92,7 @@ public sealed class Sdk2TrustedChannelTests
|
||||
private int _port;
|
||||
private int _writes;
|
||||
private int _authCalls;
|
||||
internal ConcurrentQueue<IPEndPoint?> Peers { get; } = new();
|
||||
internal OtOpcUaSdkServer Server { get; private set; } = null!;
|
||||
internal int Writes => Volatile.Read(ref _writes);
|
||||
internal int AuthCalls => Volatile.Read(ref _authCalls);
|
||||
@@ -204,6 +217,12 @@ public sealed class Sdk2TrustedChannelTests
|
||||
new Dictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>
|
||||
{ [new(NodeAclScopeKind.RawDevice, "device")] = [new([NodeAclRealms.Raw, "Plant", "Device"])] });
|
||||
|
||||
public Task<OpcUaUserAuthResult> AuthenticateUserNameAsync(string username, string password, IPEndPoint? peer, CancellationToken ct)
|
||||
{
|
||||
Peers.Enqueue(peer is null ? null : new IPEndPoint(peer.Address, peer.Port));
|
||||
return AuthenticateUserNameAsync(username, password, ct);
|
||||
}
|
||||
|
||||
public Task<OpcUaUserAuthResult> AuthenticateUserNameAsync(string username, string password, CancellationToken ct)
|
||||
{
|
||||
Interlocked.Increment(ref _authCalls);
|
||||
|
||||
+2
-3
@@ -28,7 +28,6 @@ using LocalizedText = Opc.Ua.LocalizedText;
|
||||
using Proto = ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
using UaNodeId = Opc.Ua.NodeId;
|
||||
using Variant = Opc.Ua.Variant;
|
||||
using VariantCollection = Opc.Ua.VariantCollection;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
|
||||
|
||||
@@ -505,9 +504,9 @@ public sealed class NativeAlarmRestatementEndToEndTests : RuntimeActorTestBase
|
||||
}
|
||||
|
||||
/// <summary>An upstream A&C condition event, in the driver's select-clause order.</summary>
|
||||
private static VariantCollection ConditionFields(bool active, bool acked) =>
|
||||
private static Variant[] ConditionFields(bool active, bool acked) =>
|
||||
[
|
||||
new Variant(new byte[] { 1 }), // EventId
|
||||
new Variant(Opc.Ua.ByteString.From(new byte[] { 1 })), // EventId
|
||||
new Variant(new UaNodeId("TripAlarmType", 3)), // EventType
|
||||
new Variant(UaNodeId.Parse(UpstreamSource)), // SourceNode
|
||||
new Variant(new LocalizedText("tripped")), // Message
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
using Akka.Actor;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Opc.Ua.Server;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
|
||||
@@ -14,7 +13,7 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa;
|
||||
|
||||
/// <summary>
|
||||
/// Task 60 / #81 — verifies the full path from cluster redundancy state to OPC UA
|
||||
/// <c>Server.ServiceLevel</c> visible on the wire. Boots a real <see cref="StandardServer"/>,
|
||||
/// <c>Server.ServiceLevel</c> visible on the wire. Boots a real <see cref="OtOpcUaSdkServer"/>,
|
||||
/// wires <see cref="SdkServiceLevelPublisher"/> into a <see cref="DeferredServiceLevelPublisher"/>
|
||||
/// (the production binding pattern), spawns <see cref="OpcUaPublishActor"/> against the
|
||||
/// deferred publisher, and sends a <see cref="RedundancyStateChanged"/> snapshot. Asserts
|
||||
@@ -35,7 +34,7 @@ public sealed class ServiceLevelEndToEndTests : RuntimeActorTestBase
|
||||
port => new OpcUaApplicationHost(
|
||||
BuildOptions("PrimaryLeader", pkiRoot, port),
|
||||
NullLogger<OpcUaApplicationHost>.Instance),
|
||||
() => new StandardServer(), Ct);
|
||||
() => new OtOpcUaSdkServer(), Ct);
|
||||
await using var host = started.Host;
|
||||
var server = started.Server;
|
||||
|
||||
@@ -58,7 +57,7 @@ public sealed class ServiceLevelEndToEndTests : RuntimeActorTestBase
|
||||
CorrelationId: CorrelationId.NewId()));
|
||||
|
||||
AwaitAssertion(() =>
|
||||
server.CurrentInstance.ServerObject.ServiceLevel.Value.ShouldBe((byte)240));
|
||||
server.CurrentInstance.ServerObject.ServiceLevel.ShouldNotBeNull().Value.ShouldBe((byte)240));
|
||||
}
|
||||
finally
|
||||
{
|
||||
@@ -77,7 +76,7 @@ public sealed class ServiceLevelEndToEndTests : RuntimeActorTestBase
|
||||
port => new OpcUaApplicationHost(
|
||||
BuildOptions("Secondary", pkiRoot, port),
|
||||
NullLogger<OpcUaApplicationHost>.Instance),
|
||||
() => new StandardServer(), Ct);
|
||||
() => new OtOpcUaSdkServer(), Ct);
|
||||
await using var host = started.Host;
|
||||
var server = started.Server;
|
||||
|
||||
@@ -100,7 +99,7 @@ public sealed class ServiceLevelEndToEndTests : RuntimeActorTestBase
|
||||
CorrelationId: CorrelationId.NewId()));
|
||||
|
||||
AwaitAssertion(() =>
|
||||
server.CurrentInstance.ServerObject.ServiceLevel.Value.ShouldBe((byte)100));
|
||||
server.CurrentInstance.ServerObject.ServiceLevel.ShouldNotBeNull().Value.ShouldBe((byte)100));
|
||||
}
|
||||
finally
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user