diff --git a/docs/drivers/MTConnect.md b/docs/drivers/MTConnect.md index aa6d614e..42cf80ca 100644 --- a/docs/drivers/MTConnect.md +++ b/docs/drivers/MTConnect.md @@ -73,7 +73,7 @@ below). ## Capability surface -`MTConnectDriver : IDriver, IReadable, IWritable, ISubscribable, ITagDiscovery, IHostConnectivityProbe, IRediscoverable` +`MTConnectDriver : IDriver, IReadable, IWritable, ISubscribable, ITagDiscovery, IHostConnectivityProbe, IRediscoverable, IDriverDiagnostics` (`src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs`). `IWritable` landed 2026-08-05 (`b29623cf`), against the Agent's `AllowPut` @@ -87,11 +87,12 @@ request/response handshake — is a different mechanism and remains deferred; se | Capability | Path | Notes | |------------|------|-------| | `ITagDiscovery` | `DiscoverAsync` — streams `/probe`'s Device→Component→DataItem tree into the address-space builder | `SupportsOnlineDiscovery = true`, `RediscoverPolicy = Once` | -| `IReadable` | `ReadAsync` → one `/current` per call | **Not the production data path** — see below | +| `IReadable` | `ReadAsync` → the live `/sample` pump's index while it is delivering and healthy, else one `/current` per call | **Not the production data path** — see below | | `IWritable` | `WriteAsync` → `POST {agent}/{device}?{dataItemId}={value}` on the Agent's `AllowPut` surface | **Off by default** — needs `Write.Enabled` **and** the Agent's `AllowPut`; CONDITION items are refused | | `ISubscribable` | `SubscribeAsync`/`OnDataChange` — the shared `/sample` long-poll pump | **The production data path** | -| `IHostConnectivityProbe` | periodic `/probe` under `Probe.*` | No consumer wires it today — see [Known limitations](#known-limitations) | +| `IHostConnectivityProbe` | periodic headers-only `/probe` ping under `Probe.*` — the status is read, the device model is neither downloaded in full nor parsed (review 2026-10-21 T-7; it used to parse the whole model every tick). Trade-off: a 200 from something that is not an Agent (a proxy's HTML page) now reads as reachable; the data paths still fail on it | Consumed — feeds `/hosts` (see [Known limitations](#known-limitations)) | | `IRediscoverable` | watches the Agent's `Header.instanceId` | No consumer wires it today — see [Known limitations](#known-limitations) | +| `IDriverDiagnostics` | `GetDiagnostics()` — `/sample` effective count, lag, saturated chunks, count raises | Shown on `/hosts` (review 2026-10-21 T-5) | | `IAlarmSource` | CONDITION observations off the same `/sample` pump → Part 9 transitions keyed by the alarm tag's RawPath | Opt-in per raw tag (`alarm` object) — see [CONDITION alarms](#condition-alarms-626) | ### Feature status at a glance (#613) @@ -113,6 +114,25 @@ subscribes once and lives off `OnDataChange`. `ReadAsync` exists for the Client CLI (`... read -n ...`) and for the unit/integration suites; it is correct and tested, just not on the hot path. +**Read path (review 2026-10-21 T-4 / T-2).** While the current-generation `/sample` pump is running, has +delivered since it started (a chunk, heartbeat or re-baseline) and its stream is not degraded, a read is answered +from the pump's observation index — **read only**; the pump stays its single writer — with no Agent round trip. +Otherwise (no subscription, a pump still catching up after a resubscribe, a degraded stream) it falls back to one +`/current`, whose per-batch index is now built over a tag map frozen once per binding. **Stale window:** a +stream that stalls without failing (no chunk, no heartbeat) still serves Good reads from the index until the client's +idle watchdog fires (2 × `HeartbeatMs` + `RequestTimeoutMs`) — the same bounded window subscriptions already have. During +an Agent-restart re-baseline reads fall back to `/current`. A served read also clears a +read-path degradation an earlier failed `/current` left (the read path is answering). The `/current` fallback now +checks the answer's `instanceId`: a changed one (an Agent restart seen by a browse-only driver with no pump) drops the +cached `/probe` model and raises `OnRediscoveryNeeded` once per distinct id — it used to be detected only by the pump. + +**An Agent refusal is a reply (health contract).** An `MTConnectError` document answering `/current` (e.g. +`NO_DEVICE` for a wrong `DeviceName` scope) still fails the batch (`BadCommunicationError` per ref), but it is the +Agent answering, not a comms fault: the read path is not degraded by it (that would let the comms-loss watchdog +reconnect-loop the driver over a configuration problem), `LastSuccessfulRead` advances, a prior read degradation +clears, and the Agent's text is logged once per distinct refusal. An answer that is not an MTConnect document at all +(a proxy's HTML page, truncated XML) stays a comms fault. + ## Minimum deployment ```jsonc @@ -146,11 +166,11 @@ Read off `MTConnectDriverOptions` / `MTConnectDriverConfigDto` | Key | Default | Notes | |---|---|---| -| `AgentUri` | — (required) | Agent base URI; the driver appends `/probe`, `/current`, `/sample` | +| `AgentUri` | — (required) | Agent base URI; the driver appends `/probe`, `/current`, `/sample`. Any `user:pass@` (and query) is **redacted** from `HostName` (the `/hosts` key), every log line and every exception / health message (review 2026-10-21 T-8); a URI without credentials is shown unchanged | | `DeviceName` | `null` | **Double duty.** (1) Scopes requests to `{AgentUri}/{DeviceName}/...`; `null` = whole Agent. (2) The write path's fallback device — the device a write is addressed to when the probe model does not say which device declared the DataItem (see [Write-back](#write-back--observation-injection-off-by-default)) | | `RequestTimeoutMs` | `5000` | Per-call deadline for `/probe` and `/current` | | `SampleIntervalMs` | `1000` | The Agent's `/sample?interval=` query param | -| `SampleCount` | `1000` | The Agent's `/sample?count=` query param | +| `SampleCount` | `1000` | The Agent's `/sample?count=` query param. **Caps subscription throughput** at `SampleCount / SampleIntervalMs` — see [Subscribe](#subscribe--the-sample-pump) for the adaptive raise | | `HeartbeatMs` | `10000` | The Agent's `/sample?heartbeat=` query param — also the watchdog's liveness window | | `Probe.Enabled` / `Probe.IntervalMs` / `Probe.TimeoutMs` | `true` / `5000` / `2000` | Background connectivity-probe knobs (mirrors `ModbusProbeOptions`) | | `Reconnect.MinBackoffMs` / `MaxBackoffMs` / `BackoffMultiplier` | `0` / `30000` / `2.0` | Geometric backoff after a failed request or a dropped `/sample` stream | @@ -324,6 +344,26 @@ the two need disambiguating, not just OR-ing together). On a changed `/hosts` page — see [Known limitations](#known-limitations) for what that does and does not do. +**Lag measurement and the adaptive `count` (review 2026-10-21 T-5).** Throughput is capped at `SampleCount` +observations per chunk, one chunk per `SampleIntervalMs`; a machine producing more used to fall behind the ring +buffer forever, re-baselining every buffer-length (dropping intermediate transitions) with nothing surfacing it. +Each chunk is now measured: `lag = lastSequence + 1 − nextSequence` (unknown when the Agent omits `lastSequence`) +and whether it was **saturated** (carried the full `count`). Three consecutive saturated chunks with a positive (or +unknown) lag reopen the stream at the cursor reached with `count` doubled, up to 16× `SampleCount` (max 65 536). A +raised stream that delivers nothing (an Agent refusing the larger count) steps back to the last count that delivered +and stops raising; at the ceiling a still-behind stream logs one Warning per pump. Every new pump starts from the +configured `SampleCount`. An Agent that refuses the raised count with an `MTConnectError` under HTTP 200 goes through +the ordinary unusable-answer path first — one transient Degraded blip and a backoff — before the step-back. The +driver now implements `IDriverDiagnostics`: `sampleCount` (effective), `sampleLag`, +`sampleSaturatedChunks` and `sampleCountRaises` ride the `/hosts` diagnostics bag. + +**A pump carries the session generation it was started against (review 2026-10-21 T-1).** A lifecycle restart +retires the running pump (or abandons it after its bounded teardown wait); if that pump finishes after the new +session is installed, its cursor publish, a late chunk and a late re-baseline are all dropped — checked under the +same lock the new session is installed and the generation bumped under. It used to land the OLD Agent's +`instanceId` + sequence on the new session, so the next pump read its first chunk as an Agent restart (index +cleared, one Bad flicker on every subscribed tag, a needless re-baseline). + ## Write-back — observation injection, off by default > ⚠️ **A write sets what the Agent *reports*, not what the machine does.** The value @@ -391,8 +431,26 @@ step answers a different operator question: different `level|text` wire format, and are out of scope for v1. **The Agent's refusal text is logged at Warning and never stored.** It stays out of -`DriverHealth.LastError`, which is type-only by repo-wide discipline, and a refused write -is not evidence the read path degraded — so a refusal leaves driver health untouched. +`DriverHealth.LastError`, which is type-only by repo-wide discipline. A refused write is not +evidence the read path degraded, but it **is** the Agent answering: any answer — accepted or +refused — advances `LastSuccessfulRead`, with the state following the read / `/sample` flags +(a write cannot clear either of those paths' degradation). A write that got **no** answer +(transport failure, deadline) sets **Degraded** with the stamp preserved and a type-only +`LastError` (review 2026-10-21 I-3, coordinator ruling). One success writer is shared: that +degradation is cleared by the next answered **write**, the next answered **`/current` read**, or +the next **`/sample` chunk** the current session's pump delivers (review r2 N-5: a delivered +chunk proves the Agent is alive — the analogue of OpcUaClient's good keep-alive — so it cannot +stick on a subscription-only driver). A read served from the live pump index does not clear it +by itself; the chunks behind that index do. + +**One request per device (review 2026-10-21 T-6).** The slots that survive the refusals above +go out as one `POST {agent}/{device}?id1=v1&id2=v2…` per device (the Agent's PUT surface takes +several pairs) instead of one POST per slot. A batch closes at a slot naming another device or +repeating a DataItem already in it, so two writes to one DataItem still land in request order. +An Agent refusal of a multi-slot request is retried slot by slot, so a refused DataItem costs +only its own slot (`BadCommunicationError`); a transport failure fails that request's slots. An Agent may apply +some pairs of a multi-pair PUT before refusing it, so the per-slot retry can re-send a pair that already landed — +harmless for value-setting (the Agent drops an identical repeat). **Device resolution needs no extra config.** `TagBinding.WriteTargetsByDataItemId`, built at bind time from a walk of the cached probe model, remembers which device declared diff --git a/docs/drivers/OpcUaClient.md b/docs/drivers/OpcUaClient.md index 13cd6db8..b1d2978c 100644 --- a/docs/drivers/OpcUaClient.md +++ b/docs/drivers/OpcUaClient.md @@ -25,7 +25,9 @@ out-of-process isolation. | Project | Target | Role | |---------|--------|------| | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/` | net10.0 | In-process driver — session lifetime, read / write / subscribe / alarm / history passthrough | -| `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/` | net10.0 | `IDriverBrowser` — live address-picker browse used by the AdminUI | +| `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/` | net10.0 | `IDriverBrowser` — live address-picker browse used by the AdminUI. Its side-panel `AttributesAsync` reads a variable's DataType / ValueRank / UserAccessLevel and maps them through the same `UpstreamAttributeMapping` discovery uses; it authenticates Anonymous, Username **or Certificate** through the driver's own `OpcUaUserIdentityFactory`, resolving `secret:` credentials through the host's secret store (fail-closed) — review 2026-10-21 O-5 (it was a stub, and Certificate was refused) | +| | | ⚠️ Browse Certificate auth reads `UserCertificatePath` on the **AdminUI host's** filesystem, which may differ from the driver node's; the browse session disposes the certificate it loaded when it closes. | +| `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/` | net10.0 | The SDK-typed layer the driver AND the browser share: `NamespaceMap`, `UpstreamAttributeMapping`, `OpcUaUserIdentityFactory` | | `src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/` | net10.0 | Config records + enums bound from `DriverConfig` JSON | ## Minimum deployment @@ -53,6 +55,8 @@ for every field (security policy/mode, auth type, session timeout, keep-alive, browse root, node/depth caps). `ReconnectPeriod` was **deleted** (#676 — unused since #670 removed the SDK reconnect handler, and the AdminUI no longer shows it); stored configs that still carry `reconnectPeriod` keep parsing in the factory, the Test-Connect probe and the browse picker (unknown keys are skipped), and the form drops it on the next save. +`ProbeTimeoutSeconds` (default 15, AdminUI "Probe timeout") caps the Test-Connect probe's deadline — it can shorten +the budget the host hands the probe, never extend it (review 2026-10-21 O-4; it used to be authored and consumed by nothing). ### Endpoint address: the configured host and port win (#676) @@ -70,12 +74,40 @@ A single `Session` per driver instance; subscriptions multiplex onto it. Stored session-relative `ns=N` index) so a remote namespace-table reorder across a restart doesn't silently re-point references at the wrong namespace. +**Concurrency (review 2026-10-21 O-6).** Reads (including the post-outage recovery / reconcile reads) and acks +share one gate; **writes and the discovery browse take none** (the SDK session is safe for concurrent service +calls, and the host already bounds write concurrency and serialises per tag); subscription management — data and +alarm subscribe/unsubscribe and ConditionRefresh sweeps — has its own gate. One gate used to serialise all of it, +so a write stalled for the operation timeout blocked every read and subscription create, and a discovery browse +would have blocked the data plane for its whole duration. + +**Lifecycle generation.** Every `InitializeAsync` / `ShutdownAsync` takes a new generation; the session, namespace +map and every health write land only while their generation is current. An init the host abandoned that finishes +after the next shutdown discards (closes + disposes) its session instead of installing it, and a read whose +response arrives after a shutdown does not write `Healthy` over it. A throw after the session activated closes +that local session (O-2). `ShutdownAsync` spends **one** 4 s budget on wire calls — every subscription delete +concurrently, then the close with what is left — and abandons waits the SDK does not cancel (O-9; it used to be +5 s per subscription, serially, inside the actor's `PostStop`). + ### Upstream outage and recovery (Gitea #670) The session keep-alive (`KeepAliveInterval`, default 5 s) is the liveness signal: - **Good keep-alive / any data or event notification** advances `DriverHealth.LastSuccessfulRead` — a subscription-only driver never calls `ReadAsync`, so without this its read stamp froze at connect time. + A good keep-alive with no outage latched also **clears a `Degraded` left by a failed read/write** (it used to move + only the stamp, so a subscription-only driver stayed `Degraded` until its next re-initialise). +- **Read / write / ack responses are replies.** Any service response — per-node refusals such as `BadNotWritable` + included — restores `Healthy` and advances the stamp; only a call that got **no** response (transport failure, + timeout — including an SDK request timeout surfacing as a cancellation the caller did not cause) writes `Degraded` + with the stamp preserved; acks follow the same rule. **Deliberate deviation:** a call the *caller* cancelled touches + no health — including a write the host cancelled at its own write deadline against a hung upstream; the session + keep-alive owns liveness and reports that outage itself. A reply that was in flight across a bad keep-alive advances + the stamp but does not write `Healthy` while the outage is published — the keep-alive recovery ends the outage. + An ack none of whose condition ids resolves sends nothing and logs a Warning — it does **not** throw (a deterministic + per-ack refusal must not count toward the host's AlarmAcknowledge circuit breaker, or five re-acks of one + unresolvable condition would block every genuine ack). Only a call with no reply throws; with no session an ack + fails `BadNotConnected` (before or inside the gate alike). - **Bad keep-alive** ⇒ health `Degraded` (`LastError = "keep-alive: "`, read stamp preserved), host status `Stopped`, and — **once per outage**, not per keep-alive — every subscribed ref is published `BadNoCommunication` with its last value retained and **source + server timestamp = the detection time** (live-gate F1: stamped with the @@ -231,6 +263,12 @@ initialise (the AdminUI form drops them on the next save). Pinned by > forgotten when the subscribed set changes (the host rebuilt, recreating the nodes inactive). Index keys are server-stable (`nsu=`), so a namespace reorder across a re-initialise keeps > matching. docker-dev seeds an opc-plc driver (`MAIN-opcua`); the alarm tag is > created only when `@OpcPlcAlarmSourceNodeId` is set in `docker-dev/seed/seed-clusters.sql`. +> +> **An ack that never got a reply fails (review 2026-10-21 O-8).** A transport failure on the `Acknowledge` call +> (closed channel, faulted session) — or a session lost while the ack queued — now throws with its status +> (`ServiceResultException`, e.g. `BadConnectionClosed` / `BadNotConnected`), so the host logs the ack as failed +> instead of applied. A *reply* still returns normally, including per-ack refusals such as +> `BadConditionBranchAlreadyAcked` (logged per ack): a refusal is the server answering. > This driver does **not** implement `IRediscoverable` — there is no > push-driven rediscovery signal from a remote OPC UA server in this driver. diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmConditionStateSchema.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmConditionStateSchema.cs index 38b3acf5..464cb90d 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmConditionStateSchema.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmConditionStateSchema.cs @@ -1,11 +1,13 @@ +using System.Globalization; +using System.Text.Json; using Microsoft.Data.Sqlite; namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; /// -/// DDL for the Part 9 scripted-alarm condition state: one row per alarm identity holding the -/// operator-supplied state (Enabled / Acked / Confirmed / Shelving) plus the ack/confirm audit -/// trail and comment history. +/// DDL for the Part 9 scripted-alarm condition state: one fixed-size row per alarm identity +/// holding the operator-supplied state (Enabled / Acked / Confirmed / Shelving) plus the +/// ack/confirm audit fields, and a separate append-only table holding each alarm's comment trail. /// /// /// @@ -34,14 +36,58 @@ namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; /// the last transition. Timestamps are round-trip ("O") TEXT so lexicographic order is /// chronological; nullable timestamps and audit fields are stored as NULL. /// +/// +/// The comment trail is NOT on the condition row. It used to be serialized whole into +/// comments_json, so every transition rewrote — and replicated — the alarm's entire +/// history: O(n) per transition, O(n²) over the alarm's life (review R05:4). It now lives in +/// , one row per entry keyed by (scripted_alarm_id, seq), +/// where seq is the entry's ordinal in the alarm's trail. A transition inserts only the +/// entries it appended, so the condition row stays a fixed size and a transition's write is +/// O(1) however long the history is. comments_json is kept, always written as +/// "[]", so the condition table's shape — and with it the pair's replication digest +/// for that table — does not change and an old-shape row stays readable; +/// copies any inline trail across (the alarm's next +/// save clears the column). +/// +/// +/// Why the ordinal, and not a content hash, is the key. Both halves of a redundant pair +/// run the engine and both apply every operator command, each stamping the entry with its own +/// clock. A content-derived id would keep TWO rows for every logical entry once the pair +/// converged. Keyed by ordinal, the two nodes' copies of entry n share one primary key +/// and LWW collapses them to one row — the convergence the whole-row LWW gave before, now per +/// entry. Entries go in with INSERT OR IGNORE (), so an +/// entry already held — written locally, or arrived from the peer — is never rewritten. A +/// node that MISSED a peer command keeps its own colliding entry by appending it after the +/// stored trail, with a Warning (see the store's divergence handling). +/// /// public static class AlarmConditionStateSchema { /// Table holding one condition-state row per scripted-alarm identity. public const string StateTable = "alarm_condition_state"; + /// Append-only table holding one row per comment-trail entry, keyed by alarm id + ordinal. + public const string CommentsTable = "alarm_condition_comment"; + + /// What comments_json holds once the trail lives in . + public const string EmptyInlineComments = "[]"; + /// - /// Creates the state table if it does not already exist. Idempotent. + /// The one append statement for , shared by the migration and the + /// store. OR IGNORE is what makes the trail append-only: an ordinal already held is + /// never rewritten. Parameters: @Id @Seq @Ts @User @Kind @Text. + /// + public const string InsertCommentSql = + """ + INSERT OR IGNORE INTO alarm_condition_comment + (scripted_alarm_id, seq, timestamp_utc, user_name, kind, text) + VALUES (@Id, @Seq, @Ts, @User, @Kind, @Text) + """; + + private static readonly JsonSerializerOptions InlineJsonOptions = new(JsonSerializerDefaults.Web); + + /// + /// Creates the state table and the comment table if they do not already exist. Idempotent. /// /// /// An already-open connection. ILocalDb.CreateConnection() hands out open, @@ -55,10 +101,11 @@ public static class AlarmConditionStateSchema // The four *_state columns are always written to a mapped, non-null string (Enabled/Disabled, // Acknowledged/Unacknowledged, Confirmed/Unconfirmed, Unshelved/OneShotShelved/TimedShelved), - // so they are NOT NULL. comments_json is never null either — an empty trail is the literal - // "[]". The audit user/comment/utc columns and shelving_expires_utc are NULL when the alarm - // has no such history yet. updated_at_utc carries LastTransitionUtc (there is no dedicated - // transition column). + // so they are NOT NULL. comments_json is never null either — it is the literal "[]" since the + // trail moved to alarm_condition_comment (an old-shape row may still carry an inline trail + // until the alarm's next save; MigrateInlineComments copies it without touching this row). + // The audit user/comment/utc columns and shelving_expires_utc are NULL when the alarm has no such history yet. updated_at_utc + // carries LastTransitionUtc (there is no dedicated transition column). cmd.CommandText = """ CREATE TABLE IF NOT EXISTS alarm_condition_state ( scripted_alarm_id TEXT NOT NULL PRIMARY KEY, @@ -76,7 +123,161 @@ public static class AlarmConditionStateSchema comments_json TEXT NOT NULL, updated_at_utc TEXT NOT NULL ); + + CREATE TABLE IF NOT EXISTS alarm_condition_comment ( + scripted_alarm_id TEXT NOT NULL, + seq INTEGER NOT NULL, + timestamp_utc TEXT NOT NULL, + user_name TEXT NOT NULL, + kind TEXT NOT NULL, + text TEXT NOT NULL, + PRIMARY KEY (scripted_alarm_id, seq) + ); """; cmd.ExecuteNonQuery(); } + + /// What one run did. + /// Condition rows for which at least one inline entry was copied across. + /// + /// Alarms whose inline trail is not valid JSON and was left where it is — the caller logs these, + /// so an unmigratable history is visible rather than skipped silently on every boot. + /// + public sealed record InlineCommentMigrationResult(int RowsMoved, IReadOnlyList UnparseableAlarmIds); + + /// + /// Copies every condition row's inline comments_json trail into + /// . Idempotent; it never writes the condition row itself. + /// + /// + /// + /// Run it AFTER the tables are registered for replication — it writes rows, and a + /// row written before the capture triggers exist never reaches the peer (the same + /// DDL → register → writes rule AlarmSfLegacyMigrator follows). + /// + /// + /// Copy, don't clear. Resetting comments_json here would be a full-row UPDATE + /// that re-stamps the condition row with a fresh HLC — during a pair's rolling upgrade + /// that stale-but-newly-stamped copy would beat a newer peer state (an ack made on the + /// already-upgraded node) under LWW (review m1). The column is left as it is: the store's + /// load merges an inline trail with the table by ordinal (the table wins), and the + /// engine's next save of that alarm writes "[]" with the alarm's real state. + /// + /// + /// Each entry lands at its ordinal in the inline array via , + /// so a re-run — or a row whose trail was already (partly) copied — neither duplicates nor + /// reorders anything and writes nothing for an ordinal already held. Each row copies in its + /// own transaction, so a crash leaves it fully copied or untouched. A row whose inline JSON + /// does not parse is left exactly as it is (the pre-split store could not load it either) + /// and reported in rather + /// than failing host start. + /// + /// + /// An already-open connection from ILocalDb.CreateConnection(). + /// What was copied, and which rows could not be read. + public static InlineCommentMigrationResult MigrateInlineComments(SqliteConnection connection) + { + ArgumentNullException.ThrowIfNull(connection); + + var pending = new List<(string AlarmId, string Json)>(); + using (var select = connection.CreateCommand()) + { + select.CommandText = + "SELECT scripted_alarm_id, comments_json FROM alarm_condition_state WHERE comments_json <> @Empty"; + select.Parameters.AddWithValue("@Empty", EmptyInlineComments); + using var reader = select.ExecuteReader(); + while (reader.Read()) + pending.Add((reader.GetString(0), reader.GetString(1))); + } + + var migrated = 0; + var unparseable = new List(); + foreach (var (alarmId, json) in pending) + { + IReadOnlyList entries; + try + { + entries = ParseInlineComments(json); + } + catch (JsonException) + { + unparseable.Add(alarmId); + continue; + } + + var inserted = 0; + using var tx = connection.BeginTransaction(); + for (var seq = 0; seq < entries.Count; seq++) + { + using var insert = connection.CreateCommand(); + insert.Transaction = tx; + insert.CommandText = InsertCommentSql; + insert.Parameters.AddWithValue("@Id", alarmId); + insert.Parameters.AddWithValue("@Seq", seq); + insert.Parameters.AddWithValue("@Ts", FormatTimestamp(entries[seq].TimestampUtc)); + insert.Parameters.AddWithValue("@User", entries[seq].User); + insert.Parameters.AddWithValue("@Kind", entries[seq].Kind); + insert.Parameters.AddWithValue("@Text", entries[seq].Text); + inserted += insert.ExecuteNonQuery(); + } + + tx.Commit(); + if (inserted > 0) + migrated++; + } + + return new InlineCommentMigrationResult(migrated, unparseable); + } + + /// Formats a comment timestamp for (round-trip "O" text). + /// The timestamp. + /// The round-trip text form. + public static string FormatTimestamp(DateTime value) => value.ToString("O", CultureInfo.InvariantCulture); + + /// Parses a timestamp, preserving its . + /// The round-trip text form. + /// The timestamp. + public static DateTime ParseTimestamp(string value) => + DateTime.Parse(value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind); + + /// + /// Parses a legacy inline comments_json trail exactly as the pre-split store did + /// (; null strings become empty). + /// + /// The column value. + /// The entries in trail order; empty for a blank or "[]" column. + /// The column is not a valid trail. + public static IReadOnlyList ParseInlineComments(string? json) + { + if (string.IsNullOrWhiteSpace(json)) return []; + var dtos = JsonSerializer.Deserialize>(json, InlineJsonOptions); + if (dtos is null || dtos.Count == 0) return []; + return dtos + .Select(d => new InlineComment( + d.TimestampUtc, d.User ?? string.Empty, d.Kind ?? string.Empty, d.Text ?? string.Empty)) + .ToList(); + } + + /// One comment-trail entry as stored — the persistence-side mirror of the engine's AlarmComment. + /// When the action happened. + /// Identity of the actor. + /// Classification (Acknowledge, Confirm, AddComment, …). + /// Operator-supplied or engine-generated text. + public sealed record InlineComment(DateTime TimestampUtc, string User, string Kind, string Text); + + /// The pre-split on-disk shape of one entry in comments_json. + private sealed class InlineCommentDto + { + /// When the comment was recorded. + public DateTime TimestampUtc { get; set; } + + /// Identity of the actor. + public string? User { get; set; } + + /// Classification of the entry. + public string? Kind { get; set; } + + /// Entry text. + public string? Text { get; set; } + } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectDriverOptions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectDriverOptions.cs index da204235..d46bfaf6 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectDriverOptions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectDriverOptions.cs @@ -44,6 +44,13 @@ public sealed class MTConnectDriverOptions /// The count query parameter passed to the Agent's /sample request — the /// maximum number of DataItem observations returned per chunk. Default 1000, the /// MTConnect Agent's own conventional default. + /// + /// It caps subscription throughput at SampleCount / SampleIntervalMs observations per + /// millisecond (default 1000 a second): a machine producing more falls behind the Agent's ring buffer. The + /// pump measures that lag and, when it is persistently behind, reopens the stream with the count doubled — up + /// to 16× this value, at most 65 536 — and reports the effective count and the lag on the /hosts + /// diagnostics (review 2026-10-21 T-5). Each new stream starts again from this value. + /// /// public int SampleCount { get; init; } = 1000; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectTagDefinition.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectTagDefinition.cs index aaf040b8..d2132e14 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectTagDefinition.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectTagDefinition.cs @@ -26,27 +26,17 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect; /// /// /// The DataItem's MTConnect category attribute — SAMPLE, EVENT, or -/// CONDITION. Optional metadata carried through for browse / display; not consulted by -/// the value-mapping path. Default null. -/// -/// -/// The DataItem's MTConnect type attribute (e.g. POSITION, EXECUTION). -/// Optional metadata. Default null. -/// -/// -/// The DataItem's MTConnect subType attribute (e.g. ACTUAL, COMMANDED). -/// Optional metadata. Default null. -/// -/// -/// The DataItem's MTConnect units attribute (e.g. MILLIMETER, -/// REVOLUTION/MINUTE). Optional metadata. Default null. +/// CONDITION. Read by the write path (a CONDITION is never writable). Default null. /// +/// +/// Review 2026-10-21 T-3: the MtType / MtSubType / Units members were carried here and +/// read by nothing — deleted. A stored blob or tags[] entry that still carries mtType / +/// mtSubType / units keeps loading (unknown JSON keys are skipped); the AdminUI tag editor reads +/// those keys straight off the blob for its display and type-inference advisory, not through this record. +/// public sealed record MTConnectTagDefinition( string FullName, DriverDataType DriverDataType, bool IsArray = false, int ArrayDim = 0, - string? MtCategory = null, - string? MtType = null, - string? MtSubType = null, - string? Units = null); + string? MtCategory = null); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectTagDefinitionFactory.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectTagDefinitionFactory.cs index 0458764b..5ff0a3ac 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectTagDefinitionFactory.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectTagDefinitionFactory.cs @@ -87,10 +87,7 @@ public static class MTConnectTagDefinitionFactory parsed, blob.IsArray ?? false, blob.ArrayDim ?? 0, - blob.MtCategory, - blob.MtType, - blob.MtSubType, - blob.Units); + blob.MtCategory); return true; } @@ -264,11 +261,5 @@ public static class MTConnectTagDefinitionFactory public int? ArrayDim { get; init; } /// The MTConnect DataItem category. public string? MtCategory { get; init; } - /// The MTConnect DataItem type. - public string? MtType { get; init; } - /// The MTConnect DataItem sub-type. - public string? MtSubType { get; init; } - /// The engineering units, when authored. - public string? Units { get; init; } } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/IMTConnectAgentClient.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/IMTConnectAgentClient.cs index b845e122..5ab06802 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/IMTConnectAgentClient.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/IMTConnectAgentClient.cs @@ -59,7 +59,7 @@ public sealed record MTConnectWriteOutcome(bool Accepted, string? AgentError); /// . An here would be a /// synchronous method wearing a ValueTask, and it would oblige every canned-XML fake /// to grow one too. The async unwind that a streaming client does need is already -/// covered elsewhere and is not this method's job: the enumerator +/// covered elsewhere and is not this method's job: the enumerator /// is itself and is owned by the pump that enumerates it, /// while on the client converts any read still in flight /// into an ordinary . @@ -76,10 +76,26 @@ public interface IMTConnectAgentClient : IDisposable /// The parsed device hierarchy. Task ProbeAsync(CancellationToken ct); + /// + /// A connectivity check: does the Agent answer /probe with a success status? Reads the response + /// headers only — the device model is neither downloaded in full nor parsed (R06:T-7: the connectivity + /// probe used to parse the whole /probe document every tick and throw it away). Throws exactly when + /// would for an unreachable or refusing Agent. The default delegates to + /// and discards the model. + /// + /// + /// Trade-off (review m5): a success status from something that is NOT an Agent — a proxy serving an HTML page + /// under 200 — now reads as reachable, where would have failed to parse it. The data + /// paths still report that misconfiguration (the read / stream fail to parse). + /// + /// Cancellation token (the caller's deadline). + /// A task completing when the Agent has answered with a success status. + async Task PingAsync(CancellationToken ct) => _ = await ProbeAsync(ct).ConfigureAwait(false); + /// /// Issues an Agent /current request and returns the latest observed value for every /// data item. Used both to prime the observation index at startup and to re-baseline after - /// a detected sequence gap in . + /// a detected sequence gap in . /// /// Cancellation token. /// The latest observation for every data item. @@ -113,6 +129,20 @@ public interface IMTConnectAgentClient : IDisposable /// An asynchronous sequence of streams results, one per multipart chunk. IAsyncEnumerable SampleAsync(long from, CancellationToken ct); + /// + /// As , but asking the Agent for up to + /// observations per chunk instead of the configured + /// — how the pump catches up when it falls behind an Agent + /// producing more than SampleCount / SampleIntervalMs observations a second (R06:T-5). The default + /// ignores , so a client that cannot honour it keeps working at the configured one. + /// + /// The sequence to open the stream at. + /// The per-chunk observation cap for this stream. + /// Cancellation token; the only contracted way the enumeration ends quietly. + /// The stream's chunks. + IAsyncEnumerable SampleAsync(long from, int count, CancellationToken ct) => + SampleAsync(from, ct); + /// /// Sets the Agent's observation for one DataItem, through the Agent's AllowPut /// surface (POST {agent}/{deviceName}?{dataItemId}={value}). @@ -174,6 +204,34 @@ public interface IMTConnectAgentClient : IDisposable Task WriteObservationAsync( string deviceName, string dataItemId, string value, CancellationToken cancellationToken); + /// + /// Sets several observations on ONE device in one request — the Agent's PUT surface takes any number of + /// {dataItemId}={value} query pairs (POST {agent}/{device}?a=1&b=2) — so a multi-write + /// batch costs one round trip per device instead of one per slot (R06:T-6). Same answer contract as + /// : one outcome for the whole request, an Agent refusal is a value, an + /// unreachable Agent throws. The ids must be distinct (a repeated query key's meaning is the Agent's choice). + /// The default issues the single-observation call per pair and reports the first refusal. + /// + /// The device that owns every DataItem in . + /// The (dataItemId, value) pairs, unescaped, ids distinct. + /// Cancellation token. + /// What the Agent said about the request. + async Task WriteObservationsAsync( + string deviceName, IReadOnlyList> observations, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(observations); + foreach (var (dataItemId, value) in observations) + { + var outcome = await WriteObservationAsync(deviceName, dataItemId, value, cancellationToken).ConfigureAwait(false); + if (!outcome.Accepted) + { + return outcome; + } + } + + return new MTConnectWriteOutcome(Accepted: true, AgentError: null); + } + /// /// Has the Agent's ring buffer overflowed past the sequence the caller expected next? True /// when the chunk's oldest retained sequence is strictly newer than @@ -183,7 +241,7 @@ public interface IMTConnectAgentClient : IDisposable /// /// /// is a running cursor, not the original - /// from. It equals the from passed to only + /// from. It equals the from passed to only /// for the first chunk; from the second chunk on it is the previous /// chunk's . Comparing every chunk /// against the original from instead reports a gap on every chunk once the @@ -199,7 +257,7 @@ public interface IMTConnectAgentClient : IDisposable /// a chunk the Agent was willing to send — and cppagent answers a from that has /// already fallen out of its buffer with an MTConnectError / OUT_OF_RANGE /// document served under HTTP 200, which this client surfaces as an - /// out of , never as a chunk. + /// out of , never as a chunk. /// A pump that re-baselines only on IsSequenceGap will therefore fault instead of /// recovering exactly when it has fallen furthest behind. Treat an OUT_OF_RANGE /// parse failure as a re-baseline trigger (re-prime from , diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectAgentClient.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectAgentClient.cs index 2fe22332..13e2bfab 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectAgentClient.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectAgentClient.cs @@ -17,7 +17,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect; /// The constructor opens no connection. It only composes the request URIs and builds /// its s (which themselves dial nothing until a request is issued), /// so a throwaway instance is safe to construct — the universal discovery browser's -/// CanBrowse pattern depends on that. likewise dials +/// CanBrowse pattern depends on that. likewise dials /// nothing until it is enumerated. /// /// @@ -54,7 +54,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect; /// /// /// Disposal is safe against an in-flight enumeration. Re-initialization disposes and rebuilds this client, which can happen while a pump is still -/// enumerating . Rather than letting the in-flight read fault with +/// enumerating . Rather than letting the in-flight read fault with /// an unclassified or that no /// caller is written to expect, every request runs under a token linked to an internal /// dispose token, so disposal surfaces as an ordinary @@ -79,6 +79,8 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable private readonly string _agentRoot; private readonly string _sampleUriBase; private readonly string _sampleQuerySuffix; + private readonly int _sampleCount; + private readonly int _heartbeatMs; private readonly TimeSpan _requestTimeout; private readonly TimeSpan _streamIdleTimeout; @@ -151,9 +153,10 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable _probeUri = BuildRequestUri(options, "probe"); _currentUri = BuildRequestUri(options, "current"); _sampleUriBase = BuildRequestUri(options, "sample").AbsoluteUri; - _sampleQuerySuffix = string.Create( - CultureInfo.InvariantCulture, - $"&interval={options.SampleIntervalMs}&count={options.SampleCount}&heartbeat={options.HeartbeatMs}"); + // count is appended per request: the driver may raise it for one stream when it falls behind (R06:T-5). + _sampleQuerySuffix = string.Create(CultureInfo.InvariantCulture, $"&interval={options.SampleIntervalMs}"); + _sampleCount = options.SampleCount; + _heartbeatMs = options.HeartbeatMs; _http = handler is null ? new HttpClient() : new HttpClient(handler, disposeHandler: false); _http.Timeout = _requestTimeout; @@ -179,6 +182,35 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable return MTConnectProbeParser.Parse(body); } + /// + public async Task PingAsync(CancellationToken ct) + { + ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposed) != 0, this); + + using var lifetime = CancellationTokenSource.CreateLinkedTokenSource(ct, _disposeCts.Token); + using var deadline = CancellationTokenSource.CreateLinkedTokenSource(lifetime.Token); + deadline.CancelAfter(_requestTimeout); + + using var request = new HttpRequestMessage(HttpMethod.Get, _probeUri); + HttpResponseMessage response; + try + { + // ResponseHeadersRead: the status line is the whole answer this needs. The body (the device model — + // the largest document the Agent serves) is never buffered or parsed; disposing the response drops it. + response = await _http.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, deadline.Token) + .ConfigureAwait(false); + } + catch (OperationCanceledException) when (!lifetime.IsCancellationRequested) + { + throw TimedOut(_probeUri); + } + + using (response) + { + response.EnsureSuccessStatusCode(); + } + } + /// public async Task CurrentAsync(CancellationToken ct) { @@ -195,10 +227,32 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable public async IAsyncEnumerable SampleAsync( long from, [EnumeratorCancellation] CancellationToken ct) + { + await foreach (var chunk in SampleAsync(from, count: null, ct).ConfigureAwait(false)) + { + yield return chunk; + } + } + + /// + public IAsyncEnumerable SampleAsync(long from, int count, CancellationToken ct) + { + if (count <= 0) + { + throw new ArgumentOutOfRangeException(nameof(count), count, "a /sample count must be positive"); + } + + return SampleAsync(from, (int?)count, ct); + } + + private async IAsyncEnumerable SampleAsync( + long from, + int? count, + [EnumeratorCancellation] CancellationToken ct) { ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposed) != 0, this); - var uri = BuildSampleUri(from); + var uri = BuildSampleUri(from, count); // Linked to the caller's token AND to disposal, so tearing this client down mid-enumeration // surfaces as a cancellation rather than as a raw ObjectDisposedException from the socket. @@ -226,7 +280,7 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable _ = MTConnectStreamsParser.Parse(single); throw new MTConnectStreamNotSupportedException( - $"MTConnect Agent answered the /sample request '{uri}' with a single '{response.Content.Headers.ContentType?.MediaType ?? "unknown"}' document instead of a multipart/x-mixed-replace stream. It parsed as a valid MTConnectStreams document but will never stream, so no subscription is possible; check that the URI addresses an MTConnect Agent directly and is not fronted by a proxy that buffers streaming responses."); + $"MTConnect Agent answered the /sample request '{MTConnectUriRedaction.Redact(uri)}' with a single '{response.Content.Headers.ContentType?.MediaType ?? "unknown"}' document instead of a multipart/x-mixed-replace stream. It parsed as a valid MTConnectStreams document but will never stream, so no subscription is possible; check that the URI addresses an MTConnect Agent directly and is not fronted by a proxy that buffers streaming responses."); } var reader = new MultipartStreamReader(body, boundary, _streamIdleTimeout, uri, _logger); @@ -239,7 +293,7 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable // The caller cancelling is the ONE contracted way out; anything else is reported. live.ThrowIfCancellationRequested(); - throw new MTConnectStreamEndedException(reader.EndReason, uri.AbsoluteUri, delivered); + throw new MTConnectStreamEndedException(reader.EndReason, MTConnectUriRedaction.Redact(uri), delivered); } // A frame carrying nothing but whitespace is transport filler, not a document. The @@ -258,8 +312,18 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable } /// - public async Task WriteObservationAsync( + public Task WriteObservationAsync( string deviceName, string dataItemId, string value, CancellationToken cancellationToken) + { + ArgumentException.ThrowIfNullOrWhiteSpace(dataItemId); + ArgumentNullException.ThrowIfNull(value); + + return WriteObservationsAsync(deviceName, [new KeyValuePair(dataItemId, value)], cancellationToken); + } + + /// + public async Task WriteObservationsAsync( + string deviceName, IReadOnlyList> observations, CancellationToken cancellationToken) { // Issued on _http, the unary client — the write is a short request/response // and must be bounded by RequestTimeoutMs like the other unary legs. Sending it on @@ -268,10 +332,19 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable // watchdog protects a multipart body, not a one-shot POST). ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposed) != 0, this); ArgumentException.ThrowIfNullOrWhiteSpace(deviceName); - ArgumentException.ThrowIfNullOrWhiteSpace(dataItemId); - ArgumentNullException.ThrowIfNull(value); + ArgumentNullException.ThrowIfNull(observations); + if (observations.Count == 0) + { + throw new ArgumentException("at least one observation is required", nameof(observations)); + } - var uri = BuildWriteUri(deviceName, dataItemId, value); + foreach (var (dataItemId, value) in observations) + { + ArgumentException.ThrowIfNullOrWhiteSpace(dataItemId, nameof(observations)); + ArgumentNullException.ThrowIfNull(value, nameof(observations)); + } + + var uri = BuildWriteUri(deviceName, observations); using var lifetime = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _disposeCts.Token); using var deadline = CancellationTokenSource.CreateLinkedTokenSource(lifetime.Token); @@ -326,10 +399,11 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable /// as a query pair. Every caller-supplied part is percent-encoded here — see the seam's /// contract for why escaping cannot be the caller's job. /// - private Uri BuildWriteUri(string deviceName, string dataItemId, string value) + private Uri BuildWriteUri(string deviceName, IReadOnlyList> observations) { var device = Uri.EscapeDataString(deviceName.Trim()); - var query = $"{Uri.EscapeDataString(dataItemId)}={Uri.EscapeDataString(value)}"; + var query = string.Join( + "&", observations.Select(o => $"{Uri.EscapeDataString(o.Key)}={Uri.EscapeDataString(o.Value)}")); return new Uri($"{_agentRoot}/{device}?{query}", UriKind.Absolute); } @@ -411,8 +485,10 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable } } - private Uri BuildSampleUri(long from) => - new(string.Create(CultureInfo.InvariantCulture, $"{_sampleUriBase}?from={from}{_sampleQuerySuffix}"), + private Uri BuildSampleUri(long from, int? count) => + new(string.Create( + CultureInfo.InvariantCulture, + $"{_sampleUriBase}?from={from}{_sampleQuerySuffix}&count={count ?? _sampleCount}&heartbeat={_heartbeatMs}"), UriKind.Absolute); private async Task SendAsync(Uri uri, CancellationToken ct) @@ -484,7 +560,7 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable } private TimeoutException TimedOut(Uri uri) => - new($"MTConnect Agent request to '{uri}' did not complete within {_requestTimeout.TotalMilliseconds:F0} ms."); + new($"MTConnect Agent request to '{MTConnectUriRedaction.Redact(uri)}' did not complete within {_requestTimeout.TotalMilliseconds:F0} ms."); /// /// The boundary parameter of a multipart/* response, or null when the @@ -513,7 +589,7 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable if (string.IsNullOrEmpty(boundary)) { throw new MTConnectStreamNotSupportedException( - $"MTConnect Agent answered the /sample request '{uri}' with Content-Type '{contentType.MediaType}' but no 'boundary' parameter, so the multipart stream cannot be framed."); + $"MTConnect Agent answered the /sample request '{MTConnectUriRedaction.Redact(uri)}' with Content-Type '{contentType.MediaType}' but no 'boundary' parameter, so the multipart stream cannot be framed."); } return boundary; @@ -556,7 +632,7 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable (parsed.Scheme != Uri.UriSchemeHttp && parsed.Scheme != Uri.UriSchemeHttps)) { throw new ArgumentException( - $"{nameof(MTConnectDriverOptions.AgentUri)} '{agentUri}' is not an absolute http(s) URI.", nameof(options)); + $"{nameof(MTConnectDriverOptions.AgentUri)} '{MTConnectUriRedaction.Redact(agentUri)}' is not an absolute http(s) URI.", nameof(options)); } return parsed.GetLeftPart(UriPartial.Path).TrimEnd('/'); @@ -692,7 +768,7 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable logger?.LogWarning( "MTConnect /sample stream from '{AgentUri}' sends parts with no Content-length header. Falling back to boundary-scan framing, which cannot emit a chunk until the NEXT chunk begins, so every observation is delayed by up to one Agent heartbeat. Values will read stale; this is the Agent's framing, not a driver fault.", - uri); + MTConnectUriRedaction.Redact(uri)); } private byte[]? Finish(MTConnectStreamEndReason reason) @@ -861,7 +937,7 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable if (_buffer.Length >= MaxBufferBytes) { throw new InvalidDataException( - $"MTConnect /sample stream from '{uri}' sent a part larger than {MaxBufferBytes} bytes without a boundary; refusing to buffer further."); + $"MTConnect /sample stream from '{MTConnectUriRedaction.Redact(uri)}' sent a part larger than {MaxBufferBytes} bytes without a boundary; refusing to buffer further."); } Array.Resize(ref _buffer, Math.Min(_buffer.Length * 2, MaxBufferBytes)); @@ -878,7 +954,7 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable catch (OperationCanceledException) when (!ct.IsCancellationRequested) { throw new TimeoutException( - $"MTConnect /sample stream from '{uri}' sent no data (not even a heartbeat) within {idleTimeout.TotalMilliseconds:F0} ms; treating the Agent as unreachable rather than waiting indefinitely."); + $"MTConnect /sample stream from '{MTConnectUriRedaction.Redact(uri)}' sent no data (not even a heartbeat) within {idleTimeout.TotalMilliseconds:F0} ms; treating the Agent as unreachable rather than waiting indefinitely."); } if (read <= 0) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.SampleLag.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.SampleLag.cs new file mode 100644 index 00000000..210bf153 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.SampleLag.cs @@ -0,0 +1,156 @@ +using System.Globalization; +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect; + +/// +/// Review 2026-10-21 R06:T-5 — /sample throughput is capped at +/// observations per chunk, one chunk per +/// . A machine producing more than that falls steadily behind +/// the Agent's ring buffer: every buffer-length the driver re-baselines (losing every intermediate transition) and +/// every delivered value is backlog-stale. This half measures the lag, surfaces it, and raises the per-stream +/// count so the pump can catch up. +/// +/// +/// +/// Measured per chunk. lag = lastSequence + 1 − nextSequence — the observations the Agent holds +/// that this stream has not been sent yet (unknown when the Agent omits lastSequence). A chunk is +/// saturated when it carries the full count: the Agent had more and stopped. A stream is behind +/// when consecutive chunks are saturated with a positive (or unknown) lag. +/// +/// +/// Adapted per pump. A behind stream is reopened at the cursor it reached with count doubled, up +/// to . A raised stream that then delivers nothing (an Agent that refuses the +/// larger count) steps back to the last count that delivered and stops raising for that pump. At the ceiling, +/// a stream still behind logs one Warning per pump. A new pump (a re-initialize, a resubscribe) starts again +/// from the configured — the count is an upper bound, so a +/// raised one costs nothing while the Agent has little to send. +/// +/// +/// Surfaced on the bag the /hosts page shows: the effective +/// count, the last measured lag, saturated chunks and count raises. +/// +/// +public sealed partial class MTConnectDriver : IDriverDiagnostics +{ + /// Consecutive behind chunks before the count is raised — one saturated chunk is just a burst. + internal const int BehindChunksBeforeRaise = 3; + + /// The most the per-stream count is ever raised to, relative to the configured one. + internal const int SampleCountMaxGrowth = 16; + + /// Absolute ceiling on a raised count (a stock cppagent buffer holds 131072 observations). + internal const int SampleCountAbsoluteCeiling = 65_536; + + /// The count the current pump is streaming with. Diagnostics only. + private int _effectiveSampleCount; + + /// The last measured lag (observations), or -1 when unknown. Diagnostics only. + private long _sampleLag = -1; + + /// Chunks that carried the full count. Diagnostics only. + private long _saturatedChunks; + + /// Times a pump raised its count. Diagnostics only. + private long _sampleCountRaises; + + /// The ceiling a pump started with may raise its count to. + /// The configured . + /// The ceiling — never below . + internal static int SampleCountCeiling(int configured) => + Math.Max(configured, (int)Math.Min((long)configured * SampleCountMaxGrowth, SampleCountAbsoluteCeiling)); + + /// + public IReadOnlyDictionary GetDiagnostics() + { + var bag = new Dictionary(StringComparer.Ordinal); + var count = Volatile.Read(ref _effectiveSampleCount); + if (count > 0) + { + bag["sampleCount"] = count.ToString(CultureInfo.InvariantCulture); + } + + var lag = Interlocked.Read(ref _sampleLag); + if (lag >= 0) + { + bag["sampleLag"] = lag.ToString(CultureInfo.InvariantCulture); + } + + bag["sampleSaturatedChunks"] = Interlocked.Read(ref _saturatedChunks).ToString(CultureInfo.InvariantCulture); + bag["sampleCountRaises"] = Interlocked.Read(ref _sampleCountRaises).ToString(CultureInfo.InvariantCulture); + + return bag; + } + + /// + /// Measures one delivered chunk against the count its stream was opened with and records the result. + /// + /// The chunk just applied. + /// The count the stream was opened with. + /// Whether this chunk says the stream is behind the Agent. + private bool MeasureSampleLag(MTConnectStreamsResult chunk, int sampleCount) + { + long? lag = chunk.LastSequence is { } last ? Math.Max(0, last + 1 - chunk.NextSequence) : null; + Interlocked.Exchange(ref _sampleLag, lag ?? -1); + + var saturated = chunk.Observations.Count >= sampleCount; + if (saturated) + { + Interlocked.Increment(ref _saturatedChunks); + } + + return saturated && lag is not 0; + } + + /// One pump's adaptive-count state. Only that pump's task touches it. + private sealed class SampleCountState(int configured) + { + /// The count the next stream opens with. + public int Count { get; set; } = configured; + + /// The last count a stream actually delivered at. + public int LastDelivered { get; set; } = configured; + + /// The most may be raised to; lowered to the last good count on a refusal. + public int Ceiling { get; set; } = SampleCountCeiling(configured); + + /// Whether the at-the-ceiling Warning has been logged for this pump. + public bool CeilingWarned { get; set; } + } + + /// + /// The count a behind stream should reopen with, or when it is already at its ceiling + /// (logged once per pump). + /// + private int? NextSampleCount(SampleCountState state, MTConnectDriverOptions options, long cursor) + { + var sampleCount = state.Count; + var next = (int)Math.Min((long)sampleCount * 2, state.Ceiling); + if (next > sampleCount) + { + Interlocked.Increment(ref _sampleCountRaises); + _logger.LogInformation( + "MTConnect driver {DriverInstanceId} is falling behind {AgentUri}'s /sample stream (chunks full at count {Count}, lag {Lag} observation(s)); reopening at sequence {Cursor} with count {NextCount}.", + _driverInstanceId, AgentDisplay(options), sampleCount, DescribeLag(), cursor, next); + + return next; + } + + if (!state.CeilingWarned) + { + state.CeilingWarned = true; + _logger.LogWarning( + "MTConnect driver {DriverInstanceId} is still behind {AgentUri}'s /sample stream at its count ceiling {Count} (lag {Lag} observation(s)): the machine produces more than count / SampleIntervalMs observations a second, so values are backlog-stale and the stream will re-baseline (dropping intermediate transitions) whenever it falls out of the Agent's buffer. Lower SampleIntervalMs or raise SampleCount.", + _driverInstanceId, AgentDisplay(options), sampleCount, DescribeLag()); + } + + return null; + } + + private string DescribeLag() + { + var lag = Interlocked.Read(ref _sampleLag); + return lag >= 0 ? lag.ToString(CultureInfo.InvariantCulture) : "unknown"; + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs index b6860b53..74e84efd 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs @@ -369,6 +369,25 @@ public sealed partial class MTConnectDriver /// Whether the /current read path is currently failing. See . private volatile bool _readDegraded; + /// + /// Whether the last Agent write got no reply (review I-3 / coordinator ruling: an unanswered write degrades). + /// Cleared by the next answered /current read, answered write ( / + /// ) or delivered /sample chunk of the current session + /// (, r2 N-5) — so it cannot stick on a subscription-only driver. A read served + /// from the live index does not clear it directly; the chunks behind that index do. + /// + private volatile bool _writeDegraded; + + /// The last /current refusal text logged, so a persistent refusal warns once, not per read. + private string? _lastReadRefusal; + + /// + /// Whether the current-generation /sample pump has delivered since it started (a chunk, a heartbeat or + /// a re-baseline) — the precondition, with clear, for + /// to answer from the pump's index instead of a /current (R06:T-4). + /// + private volatile bool _pumpLive; + /// /// Consecutive /sample reconnect attempts that have not yet been vindicated by /// a delivered chunk — the input to . Reset by any stream that @@ -697,7 +716,13 @@ public sealed partial class MTConnectDriver // One read of the session reference: the client used below is the one that was live at this // instant, never a half-swapped pairing (see the _session remarks). - var client = Volatile.Read(ref _session)?.Client; + IMTConnectAgentClient? client; + long readGeneration; + lock (_subscriptionLock) + { + client = Volatile.Read(ref _session)?.Client; + readGeneration = _sessionGeneration; + } var options = Volatile.Read(ref _options); // Read once, for the same reason the session is: every reference in this batch must be @@ -713,14 +738,38 @@ public sealed partial class MTConnectDriver return BadBatch(fullReferences.Count, BadNotConnected); } + // R06:T-4 — while the /sample pump is live and healthy its index IS the Agent's current state (to within + // one chunk), so a read is answered from it: READ only — the pump stays the index's single writer, which + // is all the single-writer rule below forbids. /current stays the fallback for every other case (no + // subscription, a stream not yet delivering, a degraded stream). + if (TryReadFromLivePump(fullReferences, binding) is { } live) + { + return live; + } + try { var current = await BoundedAsync(client.CurrentAsync, "/current", options.RequestTimeoutMs, cancellationToken) .ConfigureAwait(false); + // R06:T-2 — Agent-restart detection used to live only on the pump, so a driver with no subscription + // (browse-only) kept serving a cached /probe model of a process that no longer existed and never raised + // rediscovery. Only while the session this read used is still installed: a read that raced a + // re-initialise against a different Agent would otherwise announce a "restart" the operator caused. + if (current.InstanceId != Interlocked.Read(ref _announcedInstanceId) + && IsCurrentSessionGeneration(readGeneration)) + { + AnnounceAgentRestart(current.InstanceId, options); + + // Review m6: the session's cursor still named the dead instance, so the next pump would open there and + // spend its first chunk on a clear + re-baseline (a Bad flicker). Move it to the new process — through + // the same generation-checked publish the pump uses, so it can only land on the session this read used. + PublishAgentCursor(readGeneration, current.InstanceId, current.NextSequence); + } + // Built from the SAME authored tags as the shared index, so the authored-vs-unknown // distinction and every coercion behave identically — it is a snapshot, not a variant. - var snapshot = new MTConnectObservationIndex(binding.IndexTags); + var snapshot = MTConnectObservationIndex.OverFrozenTags(binding.IndexTagMap); snapshot.Apply(current, baseline: true); var results = new DataValueSnapshot[fullReferences.Count]; @@ -732,8 +781,9 @@ public sealed partial class MTConnectDriver results[i] = snapshot.Get(binding.ToDataItemId(fullReferences[i])); } - _readDegraded = false; - PublishHealthy(DateTime.UtcNow); + // m7: the next refusal after a recovery is news again. + Volatile.Write(ref _lastReadRefusal, null); + RecordReadAnswered(DateTime.UtcNow); return results; } @@ -744,6 +794,22 @@ public sealed partial class MTConnectDriver // handled below as the Agent failure it is. throw; } + catch (InvalidDataException ex) when (MTConnectXml.IsAgentRefusal(ex)) + { + // The health contract: an MTConnectError document is the Agent REFUSING the request (a bad device + // scope, an unsupported request) — it answered. The batch is Bad, but this is not a comms fault: the + // read path is not degraded by it (that would reconnect-loop a driver over a config problem), the stamp + // advances, and the Agent's text is logged once per distinct refusal. + RecordReadAnswered(DateTime.UtcNow); + if (!string.Equals(Interlocked.Exchange(ref _lastReadRefusal, ex.Message), ex.Message, StringComparison.Ordinal)) + { + _logger.LogWarning( + "MTConnect driver {DriverInstanceId}: {AgentUri} refused /current — {AgentError}. Every read is reported Bad until the Agent accepts the request; the driver stays connected.", + _driverInstanceId, AgentDisplay(options), ex.Message); + } + + return BadBatch(fullReferences.Count, BadCommunicationError); + } catch (Exception ex) { DegradeAfterReadFailure(ex); @@ -752,6 +818,56 @@ public sealed partial class MTConnectDriver } } + /// + /// Answers a read batch from the live pump's index, or returns when that index cannot be + /// trusted as the Agent's current state: no pump, a pump bound to a retired session, one that has not yet + /// delivered, or a degraded stream. Touches no health — the pump owns the stream's. + /// + /// + /// Stale window (review m1). A stream that stalls WITHOUT failing (no chunk, no heartbeat) keeps + /// _streamDegraded clear until the client's idle watchdog fires (2 × HeartbeatMs + + /// RequestTimeoutMs); for that bounded window reads answer Good from an index nothing is updating, where a + /// /current would have noticed. Subscriptions have the identical exposure; the window is the same one + /// the watchdog already accepts for the data plane. + /// + /// The batch. + /// The tag table the batch is resolved against. + /// One snapshot per reference in order, or to fall back to /current. + private DataValueSnapshot[]? TryReadFromLivePump(IReadOnlyList fullReferences, TagBinding binding) + { + lock (_subscriptionLock) + { + if (_pumpTask is not { IsCompleted: false } || _pumpGeneration != _sessionGeneration) + { + return null; + } + } + + if (!_pumpLive || _streamDegraded) + { + return null; + } + + var index = ObservationIndex; + var results = new DataValueSnapshot[fullReferences.Count]; + for (var i = 0; i < results.Length; i++) + { + results[i] = index.Get(binding.ToDataItemId(fullReferences[i])); + } + + // The read path is answering (from a stream that is delivering), so a degradation a failed /current left + // behind is no longer true of it. The stamp stays the pump's — this read heard nothing from the Agent itself. + // Clears only the READ degradation: an index read is derived from the stream's evidence — the chunks that fed the + // index are what clear an unanswered write (TryMarkStreamDelivered, r2 N-5). + if (_readDegraded) + { + _readDegraded = false; + PublishHealthy(ReadHealth().LastSuccessfulRead ?? DateTime.UtcNow); + } + + return results; + } + // ---- IWritable: observation injection through the Agent's AllowPut surface ---- /// @@ -778,9 +894,11 @@ public sealed partial class MTConnectDriver // form ⇒ BadTypeMismatch; and only then the Agent, whose refusal or whose // unreachability ⇒ BadCommunicationError. // The Agent's refusal text is logged, never stored. It goes to a Warning line and - // nowhere near , which is type-only by repo-wide - // discipline — and a refused write is not evidence the read path degraded, so health is - // left exactly as it is (unlike ). + // nowhere near LastError, which is type-only by repo-wide discipline. A refused write is + // not evidence the read path degraded — but it IS the Agent answering, so any answer + // (accepted or refused) advances LastSuccessfulRead with the state following the read / + // stream flags; a write that got no answer degrades (stamp preserved) until the next answered read or write. + // Slots that survive the local refusals go out as ONE request per device (R06:T-6). // is null — a caller bug. public async Task> WriteAsync( IReadOnlyList writes, CancellationToken cancellationToken) @@ -801,6 +919,9 @@ public sealed partial class MTConnectDriver var results = new WriteResult[writes.Count]; + // Pass 1 — every local refusal, slot by slot, in the fixed order (see above). What survives is a pending + // Agent write. + var pending = new List(writes.Count); for (var i = 0; i < writes.Count; i++) { var request = writes[i]; @@ -836,50 +957,182 @@ public sealed partial class MTConnectDriver continue; } + pending.Add(new PendingWrite(i, target.DeviceName, dataItemId, text)); + } + + if (pending.Count == 0) + { + return results; + } + + // Pass 2 — R06:T-6: the Agent's PUT surface takes several id=value pairs per device, so the pending slots go + // out as ONE request per device (it used to be one POST per slot). Request order is kept: a batch closes at + // the first slot naming a different device or repeating an id already in it, so two writes to one DataItem + // land in the order the caller sent them. + var batches = new List>(); + foreach (var write in pending) + { + var open = batches.Count > 0 ? batches[^1] : null; + if (open is null + || !string.Equals(open[0].DeviceName, write.DeviceName, StringComparison.Ordinal) + || open.Exists(w => string.Equals(w.DataItemId, write.DataItemId, StringComparison.Ordinal))) + { + batches.Add([write]); + continue; + } + + open.Add(write); + } + + for (var b = 0; b < batches.Count; b++) + { + var batch = batches[b]; try { - var outcome = await client - .WriteObservationAsync(target.DeviceName, dataItemId, text, cancellationToken) - .ConfigureAwait(false); - - if (outcome.Accepted) - { - results[i] = new WriteResult(0u); - continue; - } - - _logger.LogWarning( - "MTConnect driver {DriverInstanceId}: {AgentUri} refused the write of '{Value}' to DataItem '{DataItemId}' on device '{DeviceName}' — {AgentError}. Check the Agent's AllowPut / AllowPutFrom settings and that the DataItem exists.", - _driverInstanceId, options.AgentUri, text, dataItemId, target.DeviceName, outcome.AgentError); - - results[i] = new WriteResult(BadCommunicationError); + await WriteBatchAsync(client!, options, batch, results, cancellationToken).ConfigureAwait(false); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { - // Caller-cancelled: fill the remainder rather than throwing, so the batch shape holds. - for (var j = i; j < writes.Count; j++) + // Caller-cancelled: fill every slot not yet answered rather than throwing, so the batch shape holds + // and a slot that already landed keeps its real outcome. + for (var r = b; r < batches.Count; r++) { - results[j] = new WriteResult(BadCommunicationError); + foreach (var write in batches[r]) + { + results[write.Slot] ??= new WriteResult(BadCommunicationError); + } } return results; } - catch (Exception ex) - { - // A transport failure, a blown per-call deadline, or a disposed client. Logged with the - // exception (its TYPE is the diagnosis) and reported per slot; health is untouched. - _logger.LogWarning( - ex, - "MTConnect driver {DriverInstanceId} could not write DataItem '{DataItemId}' on device '{DeviceName}' to {AgentUri}.", - _driverInstanceId, dataItemId, target.DeviceName, options.AgentUri); - - results[i] = new WriteResult(BadCommunicationError); - } } return results; } + /// One slot that survived the local refusals, waiting for the Agent. + private sealed record PendingWrite(int Slot, string DeviceName, string DataItemId, string Text); + + /// + /// Sends one same-device batch and maps the Agent's answer onto its slots. Accepted ⇒ every slot Good. Refused + /// ⇒ a single-slot batch is that slot's BadCommunicationError (the Agent's text logged); a multi-slot + /// batch is retried slot by slot so one refused DataItem cannot fail the others — the batch contract above. A + /// transport failure fails the batch's slots. Caller cancellation propagates to . + /// + private async Task WriteBatchAsync( + IMTConnectAgentClient client, + MTConnectDriverOptions options, + List batch, + WriteResult[] results, + CancellationToken cancellationToken) + { + var device = batch[0].DeviceName; + MTConnectWriteOutcome outcome; + try + { + outcome = batch.Count == 1 + ? await client.WriteObservationAsync(device, batch[0].DataItemId, batch[0].Text, cancellationToken) + .ConfigureAwait(false) + : await client.WriteObservationsAsync( + device, + [.. batch.Select(w => new KeyValuePair(w.DataItemId, w.Text))], + cancellationToken) + .ConfigureAwait(false); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + // A transport failure, a blown per-call deadline, or a disposed client — no answer arrived. Logged with + // the exception (its TYPE is the diagnosis), reported per slot, and Degraded with the stamp preserved + // (RecordWriteNoReply) until the next answered read or write. + _logger.LogWarning( + ex, + "MTConnect driver {DriverInstanceId} could not write {Count} DataItem(s) ({DataItemIds}) on device '{DeviceName}' to {AgentUri}.", + _driverInstanceId, batch.Count, string.Join(", ", batch.Select(w => w.DataItemId)), device, AgentDisplay(options)); + + foreach (var write in batch) + { + results[write.Slot] = new WriteResult(BadCommunicationError); + } + + RecordWriteNoReply(options, ex); + + return; + } + + // An answer — accepted or refused — is the Agent talking (the health contract: a refusal is a reply). + RecordWriteAnswered(); + + if (outcome.Accepted) + { + foreach (var write in batch) + { + results[write.Slot] = new WriteResult(0u); + } + + return; + } + + if (batch.Count == 1) + { + _logger.LogWarning( + "MTConnect driver {DriverInstanceId}: {AgentUri} refused the write of '{Value}' to DataItem '{DataItemId}' on device '{DeviceName}' — {AgentError}. Check the Agent's AllowPut / AllowPutFrom settings and that the DataItem exists.", + _driverInstanceId, AgentDisplay(options), batch[0].Text, batch[0].DataItemId, device, outcome.AgentError); + results[batch[0].Slot] = new WriteResult(BadCommunicationError); + + return; + } + + // The Agent refused the request as a whole: find out which slot(s) it objects to, one by one. (Review m4: + // an Agent may have applied some pairs before refusing — cppagent collects put errors per key — so the retry + // can re-send a pair that already landed. Harmless for value-setting: the Agent's duplicate filter drops an + // identical repeat.) + foreach (var write in batch) + { + await WriteBatchAsync(client, options, [write], results, cancellationToken).ConfigureAwait(false); + } + } + + /// + /// A write got an answer from the Agent: advances, a prior + /// unanswered-write degradation clears, and the state follows the read / stream flags (a write cannot clear + /// either of THOSE paths' degradation, but it is evidence the Agent is reachable). + /// + private void RecordWriteAnswered() + { + _writeDegraded = false; + if (ReadHealth().State is DriverState.Healthy or DriverState.Degraded) + { + PublishHealthy(DateTime.UtcNow); + } + } + + /// + /// A write got NO answer (transport failure, deadline): Degraded with the stamp preserved — the health contract + /// for a comms fault. Type-only text. Cleared by the next answered read or write. + /// + private void RecordWriteNoReply(MTConnectDriverOptions options, Exception ex) + { + _writeDegraded = true; + Degrade($"MTConnect write to {AgentDisplay(options)} got no reply ({ex.GetType().Name})."); + } + + /// + /// The ONE success writer the read path and the write path share (review I-3): an answered read clears the + /// read AND write degradations (the Agent is answering requests), then publishes Healthy unless the + /// /sample stream is still failing. + /// + /// The stamp to publish. + private void RecordReadAnswered(DateTime at) + { + _readDegraded = false; + _writeDegraded = false; + PublishHealthy(at); + } + /// /// Renders a value to the invariant string a PUT query carries, or reports that it has no such /// form. @@ -1195,7 +1448,7 @@ public sealed partial class MTConnectDriver // downstream). Its data items are unreachable either way. _logger.LogWarning( "MTConnect driver {DriverInstanceId} skipped a device in {AgentUri}'s /probe model that declares neither a name nor an id; it cannot be given a browse path.", - _driverInstanceId, options.AgentUri); + _driverInstanceId, AgentDisplay(options)); continue; } @@ -1213,7 +1466,7 @@ public sealed partial class MTConnectDriver { _logger.LogWarning( "MTConnect driver {DriverInstanceId} is scoped to device '{DeviceScope}', but {AgentUri}'s /probe model declares no device with that name or id ({DeclaredCount} declared). The browse is empty because the scope matched nothing, not because the Agent has nothing.", - _driverInstanceId, deviceScope, options.AgentUri, model.Devices.Count); + _driverInstanceId, deviceScope, AgentDisplay(options), model.Devices.Count); return; } @@ -1223,7 +1476,7 @@ public sealed partial class MTConnectDriver _driverInstanceId, devices, variables, - options.AgentUri, + AgentDisplay(options), deviceScope is null or "" ? string.Empty : $" (scoped to '{deviceScope}')"); } @@ -1393,7 +1646,33 @@ public sealed partial class MTConnectDriver /// host:port convention: several MTConnect drivers in one server disambiguate on the /// Admin dashboard by endpoint rather than by driver-instance id. /// - public string HostName => Volatile.Read(ref _options).AgentUri; + public string HostName => AgentDisplay(Volatile.Read(ref _options)); + + /// + /// R06:T-8 — the redacted display form of ' AgentUri (no user:pass@), + /// used for , every log line and every message that can reach health text. Computed once + /// per options document and cached by reference. + /// + /// The options whose Agent is being named. + /// The display form; identical to the configured value when it carries no credentials. + private string AgentDisplay(MTConnectDriverOptions options) + { + var cached = Volatile.Read(ref _agentDisplay); + if (cached is not null && ReferenceEquals(cached.Options, options)) + { + return cached.Display; + } + + var fresh = new AgentDisplayCache(options, MTConnectUriRedaction.Redact(options.AgentUri)); + Volatile.Write(ref _agentDisplay, fresh); + + return fresh.Display; + } + + /// The last computed , keyed by the options instance it was computed for. + private AgentDisplayCache? _agentDisplay; + + private sealed record AgentDisplayCache(MTConnectDriverOptions Options, string Display); /// // One entry, always — an Agent is a single host, and the driver knows of it whether or not @@ -1457,9 +1736,9 @@ public sealed partial class MTConnectDriver /// free, but it only exists while something is subscribed — an instance with no /// subscriptions would report indefinitely, and host /// reachability would become a function of OPC UA client behaviour. A dedicated tick is - /// the one signal that is true whether or not anyone is watching. Its cost is a parse of - /// the device model per interval, which is why - /// exists. + /// the one signal that is true whether or not anyone is watching. Its cost used to be a parse + /// of the whole device model per interval; it is now one request whose response headers are + /// read and whose body is dropped (, R06:T-7). /// /// /// It waits first, then probes. has just completed a @@ -1499,8 +1778,8 @@ public sealed partial class MTConnectDriver using var deadline = CancellationTokenSource.CreateLinkedTokenSource(ct); deadline.CancelAfter(options.Probe.Timeout); - // The answer is deliberately discarded — see the remarks. - _ = await client.ProbeAsync(deadline.Token).ConfigureAwait(false); + // Headers only — the device model is neither downloaded in full nor parsed (R06:T-7). + await client.PingAsync(deadline.Token).ConfigureAwait(false); reachable = true; } catch (OperationCanceledException) when (ct.IsCancellationRequested) @@ -1516,7 +1795,7 @@ public sealed partial class MTConnectDriver _logger.LogDebug( ex, "MTConnect driver {DriverInstanceId} could not reach {AgentUri} on its connectivity probe.", - _driverInstanceId, options.AgentUri); + _driverInstanceId, AgentDisplay(options)); } TransitionHostTo(reachable ? HostState.Running : HostState.Stopped); @@ -1527,7 +1806,7 @@ public sealed partial class MTConnectDriver _logger.LogError( ex, "MTConnect driver {DriverInstanceId} stopped its connectivity probe on an unexpected fault; host status for {AgentUri} will not be updated until the driver is re-initialized.", - _driverInstanceId, options.AgentUri); + _driverInstanceId, AgentDisplay(options)); } } @@ -1695,7 +1974,7 @@ public sealed partial class MTConnectDriver _logger.LogInformation( "MTConnect driver {DriverInstanceId} dropped its cached /probe device model for {AgentUri} and raised rediscovery: agent instanceId is now {InstanceId}.", - _driverInstanceId, options.AgentUri, instanceId); + _driverInstanceId, AgentDisplay(options), instanceId); try { @@ -1818,7 +2097,7 @@ public sealed partial class MTConnectDriver // that is not running; without this, the resubscribe returns a handle, the caller is // told SubscriptionEstablished, and the next successful read reports a perfectly green // driver over a subscription that can never deliver a value. - var agentUri = Volatile.Read(ref _options).AgentUri; + var agentUri = AgentDisplay(Volatile.Read(ref _options)); _streamDegraded = true; Degrade($"The MTConnect Agent at '{agentUri}' does not serve a framed /sample stream; subscriptions cannot deliver until the driver is re-initialized."); @@ -1856,9 +2135,13 @@ public sealed partial class MTConnectDriver var cts = new CancellationTokenSource(); _pumpCts = cts; _pumpGeneration = _sessionGeneration; + var generation = _sessionGeneration; + // Not live until it has actually delivered (or re-baselined): a pump revived by a resubscribe opens at a + // cursor from when the last one stopped, and until it catches up the index is that old. + _pumpLive = false; _pumpTask = Task.Run( () => RunSampleStreamAsync( - session.Client, options, session.NextSequence, session.InstanceId, republishOnStart, cts.Token), + session.Client, options, session.NextSequence, session.InstanceId, republishOnStart, generation, cts.Token), CancellationToken.None); } @@ -1894,11 +2177,14 @@ public sealed partial class MTConnectDriver long from, long instanceId, bool republishOnStart, + long generation, CancellationToken ct) { var cursor = from; var attempt = 0; _reconnectAttempts = 0; + var counts = new SampleCountState(options.SampleCount); + Volatile.Write(ref _effectiveSampleCount, counts.Count); try { @@ -1929,10 +2215,41 @@ public sealed partial class MTConnectDriver } } - var outcome = await ConsumeStreamAsync(client, options, cursor, instanceId, ct).ConfigureAwait(false); + var outcome = await ConsumeStreamAsync(client, options, cursor, instanceId, generation, counts, ct).ConfigureAwait(false); cursor = outcome.Cursor; instanceId = outcome.InstanceId; + // A pump whose session a restart replaced (retired, or abandoned after the bounded teardown wait) + // acts on no verdict: its Degrade / latch would land on the NEW session's health (R06:T-1). + if (!IsCurrentSessionGeneration(generation)) + { + return; + } + + // R06:T-5 — the adaptive count. A stream that delivered vindicates the count it was opened with; a + // RAISED count whose stream delivered nothing may be one the Agent refuses (count above its buffer), + // so step back to the last count that worked and stop raising for this pump. + if (outcome.DeliveredAny) + { + counts.LastDelivered = counts.Count; + } + else if (counts.Count > counts.LastDelivered && outcome.Verdict is not StreamVerdict.Cancelled) + { + _logger.LogWarning( + outcome.Error, + "MTConnect driver {DriverInstanceId}: the /sample stream opened with the raised count {Count} delivered nothing; stepping back to {Previous} and not raising it again for this stream.", + _driverInstanceId, counts.Count, counts.LastDelivered); + counts.Count = counts.LastDelivered; + counts.Ceiling = counts.Count; + Volatile.Write(ref _effectiveSampleCount, counts.Count); + } + + if (outcome.RaiseCountTo is { } raised) + { + counts.Count = raised; + Volatile.Write(ref _effectiveSampleCount, raised); + } + switch (outcome.Verdict) { case StreamVerdict.Cancelled: @@ -1943,18 +2260,17 @@ public sealed partial class MTConnectDriver // forever, so retrying would be an unbounded hot loop against an endpoint // that will never stream until an operator changes something. Latch it, say // so loudly, and stop. A re-initialize clears the latch. - lock (_subscriptionLock) + // r1 N-4: the latch is set inside the same generation-checked section as the degrade, so a + // superseded pump can never latch Unsupported on the session that replaced it. + if (!TryDegradeStream(generation, outcome.Error!.Message, latchUnsupported: true)) { - _streamUnsupported = true; + return; } - _streamDegraded = true; - Degrade(outcome.Error!); - _logger.LogError( outcome.Error, "MTConnect driver {DriverInstanceId} cannot stream /sample from {AgentUri}: the endpoint did not answer with a framed multipart stream. Subscriptions stay registered but will receive no updates until the driver is re-initialized; check for a proxy or load balancer in front of the Agent.", - _driverInstanceId, options.AgentUri); + _driverInstanceId, AgentDisplay(options)); return; @@ -1978,8 +2294,10 @@ public sealed partial class MTConnectDriver break; default: - _streamDegraded = true; - Degrade(outcome.Error!); + if (!TryDegradeStream(generation, outcome.Error!.Message)) + { + return; + } // A stream that DELIVERED before it dropped proves the endpoint works, so // its failure starts a fresh backoff ladder. Without this the counter is @@ -1992,7 +2310,7 @@ public sealed partial class MTConnectDriver _logger.LogWarning( outcome.Error, "MTConnect driver {DriverInstanceId} lost the /sample stream from {AgentUri}; reconnecting from sequence {Cursor} (attempt {Attempt}, delivered={Delivered}).", - _driverInstanceId, options.AgentUri, cursor, attempt, outcome.DeliveredAny); + _driverInstanceId, AgentDisplay(options), cursor, attempt, outcome.DeliveredAny); break; } @@ -2005,8 +2323,7 @@ public sealed partial class MTConnectDriver "MTConnect driver {DriverInstanceId} stopped its /sample pump on an unexpected fault; subscriptions will receive no further updates until the driver is re-initialized.", _driverInstanceId); - _streamDegraded = true; - Degrade(ex); + TryDegradeStream(generation, ex.Message); } finally { @@ -2015,8 +2332,13 @@ public sealed partial class MTConnectDriver // would therefore reopen at a sequence the Agent has long since evicted: it self-heals // via OUT_OF_RANGE, but only after a wasted round trip, and a still-buffered stale // cursor replays historical observations to subscribers first. One publish per pump - // lifetime, dropped if a lifecycle change has already installed a newer session. - PublishAgentCursor(instanceId, cursor); + // lifetime, dropped if a lifecycle change has already installed a newer session (R06:T-1: the + // generation this pump was started against is what says so — the session alone cannot). + PublishAgentCursor(generation, instanceId, cursor); + if (IsCurrentSessionGeneration(generation)) + { + _pumpLive = false; + } } } @@ -2030,6 +2352,8 @@ public sealed partial class MTConnectDriver MTConnectDriverOptions options, long cursor, long instanceId, + long generation, + SampleCountState counts, CancellationToken ct) { // The RUNNING cursor the gap check is made against: the previous chunk's nextSequence, equal @@ -2043,15 +2367,26 @@ public sealed partial class MTConnectDriver // that cannot work at all. They must not share a backoff ladder — see RunSampleStreamAsync. var delivered = false; + // Consecutive chunks that said this stream is behind the Agent (R06:T-5). + var behindChunks = 0; + try { - await foreach (var chunk in client.SampleAsync(cursor, ct).ConfigureAwait(false)) + await foreach (var chunk in client.SampleAsync(cursor, counts.Count, ct).ConfigureAwait(false)) { if (chunk is null) { continue; } + // R06:T-1 — a pump whose session a lifecycle restart has replaced (it was retired, or its + // teardown wait was abandoned) must not fold a late chunk of the OLD Agent connection into + // the NEW session's index — nor read it as an Agent restart and clear that index. + if (!IsCurrentSessionGeneration(generation)) + { + return new StreamOutcome(StreamVerdict.Cancelled, expected, instanceId, delivered, null); + } + // CHECKED BEFORE THE GAP: an Agent restart changes instanceId *and* resets // sequences, so a restart usually trips IsSequenceGap too — and the two need // different handling. A gap keeps the held values (they are still this device's); @@ -2062,21 +2397,28 @@ public sealed partial class MTConnectDriver { _logger.LogWarning( "MTConnect driver {DriverInstanceId} saw {AgentUri} restart (instanceId {Previous} -> {Current}); dropping every held observation and re-baselining.", - _driverInstanceId, options.AgentUri, instanceId, chunk.InstanceId); + _driverInstanceId, AgentDisplay(options), instanceId, chunk.InstanceId); // Dropped BEFORE the re-baseline, and the subscribers are told: if the /current // then fails, holding stale values from a dead device model would be worse than // holding none. - ObservationIndex.Clear(); - ConditionTracker.Clear(); - FanOut(SubscribedReferences(), ObservationIndex); + if (!TryCaptureSessionState(generation, out var restartIndex, out var restartTracker)) + { + return new StreamOutcome(StreamVerdict.Cancelled, expected, instanceId, delivered, null); + } + + // Not live while the index is empty (review m2): reads go to /current until the re-baseline lands. + _pumpLive = false; + restartIndex.Clear(); + restartTracker.Clear(); + FanOut(SubscribedReferences(), restartIndex); // The driver's own state is consistent by this point, so it is safe to hand the // restart to arbitrary caller code — which may call straight back into // DiscoverAsync. AnnounceAgentRestart(chunk.InstanceId, options); - return await RebaselineAsync(client, options, expected, instanceId, delivered, ct) + return await RebaselineAsync(client, options, expected, instanceId, delivered, generation, ct) .ConfigureAwait(false); } @@ -2084,13 +2426,17 @@ public sealed partial class MTConnectDriver { _logger.LogWarning( "MTConnect driver {DriverInstanceId} fell out of {AgentUri}'s buffer (expected sequence {Expected}, oldest retained {FirstSequence}); re-baselining from /current.", - _driverInstanceId, options.AgentUri, expected, chunk.FirstSequence); + _driverInstanceId, AgentDisplay(options), expected, chunk.FirstSequence); - return await RebaselineAsync(client, options, expected, instanceId, delivered, ct) + return await RebaselineAsync(client, options, expected, instanceId, delivered, generation, ct) .ConfigureAwait(false); } - ApplyAndFanOut(chunk); + if (!ApplyAndFanOut(chunk, generation)) + { + return new StreamOutcome(StreamVerdict.Cancelled, expected, instanceId, delivered, null); + } + delivered = true; // EVERY chunk advances the cursor, including an observation-free heartbeat: the @@ -2099,8 +2445,27 @@ public sealed partial class MTConnectDriver // like a gap. expected = chunk.NextSequence; - _streamDegraded = false; - PublishHealthy(DateTime.UtcNow); + // Review I-1: a pump wedged inside the fan-out above can come back after a restart replaced its + // session — its "the stream delivered" must then land nowhere. + if (!TryMarkStreamDelivered(generation)) + { + return new StreamOutcome(StreamVerdict.Cancelled, expected, instanceId, delivered, null); + } + + // R06:T-5 — a stream persistently behind the Agent reopens at the cursor it reached with a larger + // count, rather than falling out of the ring buffer and re-baselining every buffer-length. + if (!MeasureSampleLag(chunk, counts.Count)) + { + behindChunks = 0; + } + else if (++behindChunks >= BehindChunksBeforeRaise) + { + behindChunks = 0; + if (NextSampleCount(counts, options, expected) is { } next) + { + return new StreamOutcome(StreamVerdict.Reopen, expected, instanceId, delivered, null, next); + } + } } // The seam contracts that cancellation is the ONLY way this enumeration ends quietly, so @@ -2114,7 +2479,7 @@ public sealed partial class MTConnectDriver instanceId, delivered, new MTConnectStreamEndedException( - MTConnectStreamEndReason.ConnectionClosed, options.AgentUri, 0)); + MTConnectStreamEndReason.ConnectionClosed, AgentDisplay(options), 0)); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { @@ -2134,9 +2499,9 @@ public sealed partial class MTConnectDriver _logger.LogWarning( ex, "MTConnect driver {DriverInstanceId} could not resume {AgentUri}'s /sample stream at sequence {Cursor} (the Agent rejected it as out of range); re-baselining from /current.", - _driverInstanceId, options.AgentUri, cursor); + _driverInstanceId, AgentDisplay(options), cursor); - return await RebaselineAsync(client, options, expected, instanceId, delivered, ct) + return await RebaselineAsync(client, options, expected, instanceId, delivered, generation, ct) .ConfigureAwait(false); } catch (Exception ex) @@ -2166,6 +2531,7 @@ public sealed partial class MTConnectDriver long fallbackCursor, long fallbackInstanceId, bool deliveredAny, + long generation, CancellationToken ct) { try @@ -2173,16 +2539,23 @@ public sealed partial class MTConnectDriver var current = await BoundedAsync(client.CurrentAsync, "/current", options.RequestTimeoutMs, ct) .ConfigureAwait(false); - PublishAgentCursor(current.InstanceId, current.NextSequence); - ApplyAndFanOut(current, baseline: true); + if (!IsCurrentSessionGeneration(generation)) + { + return new StreamOutcome( + StreamVerdict.Cancelled, fallbackCursor, fallbackInstanceId, deliveredAny, null); + } - _streamDegraded = false; - PublishHealthy(DateTime.UtcNow); + PublishAgentCursor(generation, current.InstanceId, current.NextSequence); + if (!ApplyAndFanOut(current, generation, baseline: true) || !TryMarkStreamDelivered(generation)) + { + return new StreamOutcome( + StreamVerdict.Cancelled, fallbackCursor, fallbackInstanceId, deliveredAny, null); + } _logger.LogInformation( "MTConnect driver {DriverInstanceId} re-baselined from {AgentUri}'s /current ({ObservationCount} observation(s), agent instanceId {InstanceId}) and resumes /sample from sequence {NextSequence}.", _driverInstanceId, - options.AgentUri, + AgentDisplay(options), current.Observations.Count, current.InstanceId, current.NextSequence); @@ -2257,19 +2630,29 @@ public sealed partial class MTConnectDriver /// Folds a /current snapshot or one /sample chunk into the shared index and /// publishes every reference it touched to the handles that subscribe it. /// - private void ApplyAndFanOut(MTConnectStreamsResult document, bool baseline = false) + /// The document to fold in. + /// The session generation of the pump doing it. + /// Whether it is a /current baseline. + /// (nothing applied) when the pump's session has been replaced. + private bool ApplyAndFanOut(MTConnectStreamsResult document, long generation, bool baseline = false) { - var index = ObservationIndex; + // Captured with the generation check under one lock (review m3): a stale chunk can never fold into the + // session state a restart installed after the pump's last check. + if (!TryCaptureSessionState(generation, out var index, out var tracker)) + { + return false; + } + index.Apply(document, baseline); // CONDITION observations also feed the alarm path, which keeps its own multi-condition state // (a /sample Normal for one nativeCode is not a return to normal while another code is active). - var touchedConditions = ConditionTracker.Apply(document, baseline); + var touchedConditions = tracker.Apply(document, baseline); EmitConditionTransitions(touchedConditions, resync: false); if (document.Observations is not { Count: > 0 }) { - return; + return true; } // De-duplicated in document order: one Agent document may report a data item more than once @@ -2298,6 +2681,85 @@ public sealed partial class MTConnectDriver } FanOut(touched, index); + + return true; + } + + /// + /// The observation index and condition tracker of the session names, read with + /// the generation check under — the lock installs + /// them, the session and the generation under. once that session is gone. + /// + private bool TryCaptureSessionState( + long generation, out MTConnectObservationIndex index, out MTConnectConditionTracker tracker) + { + lock (_subscriptionLock) + { + index = Volatile.Read(ref _index); + tracker = Volatile.Read(ref _conditions); + + return _sessionGeneration == generation; + } + } + + /// + /// The stream of the pump started under failed: degrade it — under the generation + /// check, in one critical section, so a superseded pump's failure never lands on the session that replaced it. + /// + /// The pump's session generation. + /// The failure (redacted text only). + /// when the pump's session has been replaced (nothing written). + /// Also latch the "endpoint cannot stream" verdict (r1 N-4). + private bool TryDegradeStream(long generation, string message, bool latchUnsupported = false) + { + // The mirror of TryMarkStreamDelivered: a superseded pump's failure is not the new session's. + lock (_subscriptionLock) + { + if (_sessionGeneration != generation) + { + return false; + } + + if (latchUnsupported) + { + _streamUnsupported = true; + } + + _streamDegraded = true; + Degrade(message); + + return true; + } + } + + /// + /// Review I-1: the stream of the pump started under delivered — clear its + /// degradation, mark it live and publish Healthy, all under the generation check and in one critical section, + /// so a superseded pump (wedged in a subscriber past its last check) can never restore a replaced session's + /// health or make the read path trust a stream that is down. Nothing here calls caller code. + /// + /// The pump's session generation. + /// when the pump's session has been replaced (nothing written). + private bool TryMarkStreamDelivered(long generation) + { + lock (_subscriptionLock) + { + if (_sessionGeneration != generation) + { + return false; + } + + _streamDegraded = false; + _pumpLive = true; + + // r2 N-5 (coordinator): a delivered chunk proves the Agent is alive — the MTConnect analogue of OpcUaClient's + // good keep-alive clearing a read/write Degraded — so it also clears an unanswered-write degradation, which + // otherwise stuck forever on a subscription-only driver. Generation-checked with the rest (above). + _writeDegraded = false; + PublishHealthy(DateTime.UtcNow); + + return true; + } } /// @@ -2488,25 +2950,51 @@ public sealed partial class MTConnectDriver /// /// The Agent instanceId last observed. /// The sequence a /sample stream should resume from. - private void PublishAgentCursor(long instanceId, long nextSequence) + /// The session generation the publishing pump was started against. + private void PublishAgentCursor(long generation, long instanceId, long nextSequence) { - while (true) + // R06:T-1 — under _subscriptionLock, the lock StartCoreAsync installs a new session AND bumps the + // generation under: inside it the two are always observed together, so a pump bound to a retired + // session can never land its cursor on the session that replaced it (a lock-free generation check + // could read the new session beside the not-yet-bumped generation). Nothing is awaited here. + lock (_subscriptionLock) { - var session = Volatile.Read(ref _session); - if (session is null || - (session.InstanceId == instanceId && session.NextSequence == nextSequence)) + if (_sessionGeneration != generation) { return; } - var updated = session with { InstanceId = instanceId, NextSequence = nextSequence }; - if (ReferenceEquals(Interlocked.CompareExchange(ref _session, updated, session), session)) + while (true) { - return; + var session = Volatile.Read(ref _session); + if (session is null || + (session.InstanceId == instanceId && session.NextSequence == nextSequence)) + { + return; + } + + // Still a compare-and-swap: InvalidateProbeModel / GetOrFetchProbeModelAsync swap the session + // lock-free, and a plain write here could undo a probe-model drop. + var updated = session with { InstanceId = instanceId, NextSequence = nextSequence }; + if (ReferenceEquals(Interlocked.CompareExchange(ref _session, updated, session), session)) + { + return; + } } } } + /// Whether is still the installed session's generation. + /// A pump's session generation. + /// once a lifecycle restart has installed a newer session. + private bool IsCurrentSessionGeneration(long generation) + { + lock (_subscriptionLock) + { + return _sessionGeneration == generation; + } + } + /// /// The /probe device model: the cached copy when warm, otherwise a fresh /// deadline-bounded /probe that re-populates the cache. This is the accessor every @@ -2715,28 +3203,32 @@ public sealed partial class MTConnectDriver // The generation bump and the session write are ONE critical section: a Subscribe landing // between them would start a pump on the OLD session stamped with the NEW generation, and // the restart's own StartSampleStreamCore would then treat it as current. + var index = new MTConnectObservationIndex(binding.IndexTags); + index.Apply(current, baseline: true); + + // the CONDITION tracker is per-session state like the index. Nothing is RAISED here — the + // lifecycle semaphore is held; the restarted pump delivers the diff as its first act. + var conditions = new MTConnectConditionTracker(); + conditions.Apply(current, baseline: true); + + // The index and the tracker are installed in the SAME critical section as the session and its + // generation (R06:T-1 / review m3): a pump that reads them under this lock with its generation still + // current gets this session's state, never the previous one's, and never the next one's. lock (_subscriptionLock) { Volatile.Write( ref _session, new AgentSession(client, probe, current.InstanceId, current.NextSequence)); _sessionGeneration++; + Volatile.Write(ref _index, index); + Volatile.Write(ref _conditions, conditions); } - var index = new MTConnectObservationIndex(binding.IndexTags); - index.Apply(current, baseline: true); - Volatile.Write(ref _index, index); - - // the CONDITION tracker is per-session state like the index. Nothing is RAISED here — the - // lifecycle semaphore is held; the restarted pump delivers the diff as its first act. - var conditions = new MTConnectConditionTracker(); - conditions.Apply(current, baseline: true); - Volatile.Write(ref _conditions, conditions); - // A fresh session has no failures behind it — including the /sample verdict, because a // re-initialize is precisely the event that means an operator changed something. _streamDegraded = false; _readDegraded = false; + _writeDegraded = false; WriteHealth(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null)); @@ -2756,7 +3248,7 @@ public sealed partial class MTConnectDriver _logger.LogInformation( "MTConnect driver {DriverInstanceId} initialized against {AgentUri}{DeviceScope}: {DeviceCount} device(s), {ObservationCount} primed observation(s), agent instanceId {InstanceId}.", _driverInstanceId, - options.AgentUri, + AgentDisplay(options), options.DeviceName is null ? string.Empty : $"/{options.DeviceName}", probe.Devices.Count, index.Count, @@ -2788,7 +3280,7 @@ public sealed partial class MTConnectDriver _logger.LogError( ex, "MTConnect driver {DriverInstanceId} failed to initialize against {AgentUri}; the driver is Faulted and serves no values.", - _driverInstanceId, options.AgentUri); + _driverInstanceId, AgentDisplay(options)); throw; } @@ -2804,8 +3296,16 @@ public sealed partial class MTConnectDriver // Retire the whole session in one write, BEFORE disposing: a lock-free reader that still // holds the old reference gets a disposed client (caught + named at each seam), never a // session that is half-torn-down. - var session = Volatile.Read(ref _session); - Volatile.Write(ref _session, null); + // The retirement bumps the session generation too, so a pump abandoned by the teardown wait can no + // longer write health, flags or index state over what the shutdown / next start writes (review I-1). + AgentSession? session; + lock (_subscriptionLock) + { + session = Volatile.Read(ref _session); + Volatile.Write(ref _session, null); + _sessionGeneration++; + } + SafeDispose(session?.Client); // A fresh index rather than Clear(): the tag table it coerces against belongs to the options @@ -3092,7 +3592,7 @@ public sealed partial class MTConnectDriver { // A bare "A task was canceled" names neither the Agent nor the request that stalled. throw new TimeoutException( - $"MTConnect {request} request for driver '{_driverInstanceId}' against '{Volatile.Read(ref _options).AgentUri}' did not complete within {timeoutMs} ms."); + $"MTConnect {request} request for driver '{_driverInstanceId}' against '{AgentDisplay(Volatile.Read(ref _options))}' did not complete within {timeoutMs} ms."); } } @@ -3129,7 +3629,7 @@ public sealed partial class MTConnectDriver _logger.LogWarning( ex, "MTConnect driver {DriverInstanceId} could not read /current from {AgentUri}; the batch is reported Bad.", - _driverInstanceId, Volatile.Read(ref _options).AgentUri); + _driverInstanceId, AgentDisplay(Volatile.Read(ref _options))); } /// Publishes without downgrading a Faulted driver. @@ -3186,7 +3686,7 @@ public sealed partial class MTConnectDriver { var current = ReadHealth(); - if (_streamDegraded || _readDegraded) + if (_streamDegraded || _readDegraded || _writeDegraded) { if (current.State != DriverState.Faulted) { @@ -3277,10 +3777,7 @@ public sealed partial class MTConnectDriver : ParseEnum(dto.DriverDataType, dto.FullName, driverInstanceId, nameof(MTConnectTagDefinition.DriverDataType)), dto.IsArray ?? false, dto.ArrayDim ?? 0, - dto.MtCategory, - dto.MtType, - dto.MtSubType, - dto.Units); + dto.MtCategory); } // ---- v3 tag binding: RawPath ↔ dataItemId, and the definitions the index coerces against ---- @@ -3549,10 +4046,7 @@ public sealed partial class MTConnectDriver inferred.DataType, inferred.IsArray, (int)(inferred.ArrayDim ?? 0), - dataItem.Category, - dataItem.Type, - dataItem.SubType, - dataItem.Units); + dataItem.Category); } /// Flattens a device model into a dataItemId → DataItem lookup; null ⇒ empty. @@ -3647,6 +4141,10 @@ public sealed partial class MTConnectDriver IReadOnlyList IndexTags, FrozenDictionary WriteTargetsByDataItemId) { + /// frozen once per binding, for the per-read index (R06:T-4). + public FrozenDictionary IndexTagMap { get; } = + MTConnectObservationIndex.FreezeTags(IndexTags); + /// /// Resolves a write reference to the DataItem id and the Agent address it writes through. /// @@ -3765,8 +4263,12 @@ public sealed partial class MTConnectDriver /// the reconnect backoff from ratcheting monotonically over a process lifetime. /// /// The failure, when there was one. + /// + /// Set on a the adaptive count asked for: the stream was persistently behind + /// the Agent, and the next one opens with this count (R06:T-5). + /// private sealed record StreamOutcome( - StreamVerdict Verdict, long Cursor, long InstanceId, bool DeliveredAny, Exception? Error); + StreamVerdict Verdict, long Cursor, long InstanceId, bool DeliveredAny, Exception? Error, int? RaiseCountTo = null); // ---- config DTOs ---- @@ -3818,12 +4320,9 @@ public sealed partial class MTConnectDriver public int? ArrayDim { get; init; } /// The MTConnect DataItem category. public string? MtCategory { get; init; } - /// The MTConnect DataItem type. - public string? MtType { get; init; } - /// The MTConnect DataItem sub-type. - public string? MtSubType { get; init; } - /// The engineering units. - public string? Units { get; init; } + + // mtType / mtSubType / units were bound here and read by nothing (review 2026-10-21 T-3); a document that + // still carries them parses — System.Text.Json skips unmapped members. } /// Background connectivity-probe knobs (consumed by the probe loop). diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDtos.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDtos.cs index 9672d455..d9992bf0 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDtos.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDtos.cs @@ -56,7 +56,7 @@ public sealed record MTConnectComponent( /// /// One MTConnect DataItem declaration from the Agent's probe response. This is pure metadata — /// it carries no observed value; values arrive separately via -/// / +/// / /// and are correlated back to a data item by . /// /// @@ -89,7 +89,7 @@ public sealed record MTConnectDataItem( /// /// Result of an Agent /current request, or of a single multipart chunk from a /// /sample stream. Produced by and -/// . +/// . /// /// /// The Agent's current instance id (its MTConnectStreams header instanceId). Changes @@ -98,7 +98,7 @@ public sealed record MTConnectDataItem( /// /// /// The header nextSequence — the sequence number to resume a subsequent -/// call from. +/// call from. /// /// /// The header firstSequence — the oldest sequence number still held in the Agent's @@ -110,11 +110,17 @@ public sealed record MTConnectDataItem( /// The observations carried in this snapshot/chunk, in Agent-supplied order. Never null; /// empty when the chunk carries no new observations (e.g. a heartbeat). /// +/// +/// The header lastSequence — the newest sequence number the Agent holds, or null when the +/// document omits it. With it measures how far a /sample consumer is behind +/// the Agent: LastSequence + 1 - NextSequence observations it has not been sent yet (R06:T-5). +/// public sealed record MTConnectStreamsResult( long InstanceId, long NextSequence, long FirstSequence, - IReadOnlyList Observations); + IReadOnlyList Observations, + long? LastSequence = null); /// /// One observed value for a single DataItem, as reported in a /current snapshot or a @@ -162,11 +168,6 @@ public sealed record MTConnectStreamsResult( /// The observation's own sampleCount attribute — the number of values a TIME_SERIES /// vector claims to carry. The index checks the parsed vector against it. null when absent. /// -/// -/// The observation's own sampleRate attribute (samples per second) on a TIME_SERIES -/// observation. Carried for the parser contract; not yet surfaced on the served node (see -/// docs/drivers/MTConnect.md). null when absent. -/// /// /// The parsed <Entry key="…"> children of a DATA_SET / TABLE observation, in /// document order, or null when the observation had none. A TABLE entry carries @@ -195,7 +196,6 @@ public sealed record MTConnectObservation( DateTime TimestampUtc, bool IsStructured = false, int? SampleCount = null, - double? SampleRate = null, MTConnectConditionDetail? Condition = null, IReadOnlyList? Entries = null, bool ResetTriggered = false, diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectObservationIndex.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectObservationIndex.cs index ec6561f3..8c463697 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectObservationIndex.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectObservationIndex.cs @@ -151,7 +151,34 @@ internal sealed class MTConnectObservationIndex TimeProvider? timeProvider = null) { _timeProvider = timeProvider ?? TimeProvider.System; + _tags = FreezeTags(tags); + } + /// Builds an index over an already-frozen tag map (see ). + private MTConnectObservationIndex(FrozenDictionary frozenTags, TimeProvider? timeProvider) + { + _timeProvider = timeProvider ?? TimeProvider.System; + _tags = frozenTags; + } + + /// + /// R06:T-4 — an index over a tag map frozen ONCE (per binding) rather than per construction: the + /// /current read fallback builds a fresh index per batch, and re-freezing a 2–5k-entry tag table on + /// every read was a measurable share of that batch. + /// + /// A map produced by . + /// A fresh, empty index coercing against . + internal static MTConnectObservationIndex OverFrozenTags(FrozenDictionary frozenTags) => + new(frozenTags, null); + + /// + /// The authored tags keyed by , last definition winning on a + /// duplicate (see the constructor), blank / null entries skipped. + /// + /// The authored tags. + /// The frozen map an index coerces against. + internal static FrozenDictionary FreezeTags(IEnumerable? tags) + { var map = new Dictionary(StringComparer.Ordinal); foreach (var tag in tags ?? []) { @@ -160,7 +187,7 @@ internal sealed class MTConnectObservationIndex map[tag.FullName] = tag; } - _tags = map.ToFrozenDictionary(StringComparer.Ordinal); + return map.ToFrozenDictionary(StringComparer.Ordinal); } /// The number of distinct dataItemIds currently indexed. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectProbeParser.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectProbeParser.cs index 4d884e1a..0a184e49 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectProbeParser.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectProbeParser.cs @@ -118,7 +118,7 @@ internal static class MTConnectProbeParser if (!MTConnectXml.IsNamed(root, "MTConnectDevices")) { - throw new InvalidDataException(MTConnectXml.DescribeUnexpectedRoot(root, "MTConnectDevices", Subject)); + throw MTConnectXml.UnexpectedRoot(root, "MTConnectDevices", Subject); } var devicesContainer = MTConnectXml.ChildrenNamed(root, "Devices").FirstOrDefault() diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamExceptions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamExceptions.cs index fe0b54a9..fa41c55a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamExceptions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamExceptions.cs @@ -7,7 +7,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect; /// /// /// Why these are exceptions rather than a normal end of enumeration. -/// is contracted to yield indefinitely until +/// is contracted to yield indefinitely until /// its token is cancelled, so a pump written to that contract has no reason to handle normal /// completion — it would either drop the subscription silently or hot-loop reconnecting on /// an enumerator that ends immediately. Returning normally would make "the data path @@ -159,3 +159,4 @@ public sealed class MTConnectStreamNotSupportedException : MTConnectStreamExcept { } } + diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamsParser.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamsParser.cs index 19b3df29..4e32e5f7 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamsParser.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamsParser.cs @@ -295,8 +295,7 @@ internal static class MTConnectStreamsParser // exactly this shape (Agents serve one under HTTP 200), and its elements — the // Agent's own account of what it objected to — are the most useful thing in the whole // response. Materialising the root is how the shared extraction can reach them. - throw new InvalidDataException( - MTConnectXml.DescribeUnexpectedRoot((XElement)XNode.ReadFrom(reader), "MTConnectStreams", Subject)); + throw MTConnectXml.UnexpectedRoot((XElement)XNode.ReadFrom(reader), "MTConnectStreams", Subject); } var observations = new List(); @@ -330,7 +329,8 @@ internal static class MTConnectStreamsParser sequences.InstanceId, sequences.NextSequence, sequences.FirstSequence, - observations); + observations, + sequences.LastSequence); } /// @@ -342,7 +342,14 @@ internal static class MTConnectStreamsParser new( MTConnectXml.RequiredLongAttribute(reader, "instanceId", HeaderOwner, Subject), MTConnectXml.RequiredLongAttribute(reader, "nextSequence", HeaderOwner, Subject), - MTConnectXml.RequiredLongAttribute(reader, "firstSequence", HeaderOwner, Subject)); + MTConnectXml.RequiredLongAttribute(reader, "firstSequence", HeaderOwner, Subject), + // Optional on purpose: only the /sample lag measurement reads it, and a document without it must + // still parse (the pump then falls back to chunk saturation alone — R06:T-5). + long.TryParse( + MTConnectXml.OptionalAttribute(reader, "lastSequence"), + NumberStyles.Integer, + CultureInfo.InvariantCulture, + out var last) ? last : null); /// /// Consumes the element the reader is positioned on, collecting any observation it is and @@ -383,7 +390,6 @@ internal static class MTConnectStreamsParser var timestampUtc = default(DateTime); string? conditionState = null; int? sampleCount = null; - double? sampleRate = null; var resetTriggered = false; var hasCount = false; List? entries = null; @@ -421,11 +427,6 @@ internal static class MTConnectStreamsParser NumberStyles.Integer, CultureInfo.InvariantCulture, out var count) && count >= 0 ? count : null; - sampleRate = double.TryParse( - MTConnectXml.OptionalAttribute(reader, "sampleRate"), - NumberStyles.Float, - CultureInfo.InvariantCulture, - out var rate) && double.IsFinite(rate) ? rate : null; } // The slot is reserved BEFORE the subtree is read, and filled after: an observation is @@ -485,7 +486,6 @@ internal static class MTConnectStreamsParser // away a perfectly well-determined state. IsStructured: !isCondition && hasElements, SampleCount: sampleCount, - SampleRate: sampleRate, Condition: condition, Entries: entries is { Count: > 0 } ? entries : null, @@ -660,7 +660,7 @@ internal static class MTConnectStreamsParser } /// The three sequence numbers read off a <Header>. - private readonly record struct HeaderSequences(long InstanceId, long NextSequence, long FirstSequence); + private readonly record struct HeaderSequences(long InstanceId, long NextSequence, long FirstSequence, long? LastSequence); /// /// Accumulates one observation's text across however many nodes the parser hands it, without diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectUriRedaction.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectUriRedaction.cs new file mode 100644 index 00000000..5cff4513 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectUriRedaction.cs @@ -0,0 +1,60 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect; + +/// +/// Review 2026-10-21 R06:T-8 — the display form of an Agent URI for every log line, exception message, +/// DriverHealth.LastError and HostName this driver produces. An AgentUri may carry +/// user:pass@; the Test-Connect probe already redacted it (MTConnectDriverProbe.RedactedDisplay) +/// while the driver itself logged and surfaced the raw string — and every request URI built from it. +/// +/// +/// A URI with no user-info (and, for the configured string, no query or fragment) is returned unchanged, +/// so the common case — and every host key derived from it — is byte-identical to what it was. One that does +/// carry user-info loses it (and, for the configured string, its query / fragment, which a reverse proxy may use +/// for a token). A configured string that is not an absolute URI but contains an @ is replaced wholesale: +/// there is no reliable way to find the credential in it. +/// +internal static class MTConnectUriRedaction +{ + /// Placeholder for a configured value that cannot be parsed and might carry a credential. + internal const string Unparseable = "(agentUri withheld: not an absolute URI)"; + + /// The display form of a configured AgentUri. + /// The configured value. + /// The value with any user-info (and query / fragment) removed. + public static string Redact(string? agentUri) + { + if (string.IsNullOrWhiteSpace(agentUri)) + { + return agentUri ?? string.Empty; + } + + if (!Uri.TryCreate(agentUri.Trim(), UriKind.Absolute, out var uri)) + { + return agentUri.Contains('@', StringComparison.Ordinal) ? Unparseable : agentUri; + } + + if (uri.UserInfo.Length == 0 && uri.Query.Length == 0 && uri.Fragment.Length == 0) + { + return agentUri; + } + + var port = uri.IsDefaultPort ? string.Empty : $":{uri.Port}"; + var path = uri.AbsolutePath == "/" ? string.Empty : uri.AbsolutePath; + + return $"{uri.Scheme}://{uri.Host}{port}{path}"; + } + + /// The display form of a request URI built from the AgentUri (its query is the request's own). + /// The request URI. + /// The URI as text, without user-info. + public static string Redact(Uri uri) + { + ArgumentNullException.ThrowIfNull(uri); + if (!uri.IsAbsoluteUri || uri.UserInfo.Length == 0) + { + return uri.ToString(); + } + + return new UriBuilder(uri) { UserName = string.Empty, Password = string.Empty }.Uri.ToString(); + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectXml.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectXml.cs index a92d376b..3497d27f 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectXml.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectXml.cs @@ -190,7 +190,44 @@ internal static class MTConnectXml /// The document's actual root element. /// The root element name the caller required. /// How to describe the document in the error message. - /// The error message describing the unexpected root. + /// + /// The exception to throw. For an MTConnectError root (the Agent REFUSED the request) it is marked + /// with — see . + /// + public static InvalidDataException UnexpectedRoot(XElement root, string expectedRootName, string subject) + { + var ex = new InvalidDataException(DescribeUnexpectedRoot(root, expectedRootName, subject)); + + // Marked (InvalidDataException is sealed, so not a subtype) so the read path can tell a refusal — the + // Agent answered — from an answer it could not parse at all. Every other handler sees the same exception. + if (IsNamed(root, "MTConnectError")) + { + ex.Data[AgentRefusalDataKey] = true; + } + + return ex; + } + + /// key marking an MTConnectError answer (see ). + internal const string AgentRefusalDataKey = "MTConnect.AgentRefusal"; + + /// + /// Whether is the Agent answering with an MTConnectError document (e.g. + /// NO_DEVICE, INVALID_REQUEST) — a refusal, as opposed to an answer that could not be parsed. + /// + /// The exception a parse raised. + /// for a refusal. + public static bool IsAgentRefusal(Exception ex) => + ex is InvalidDataException && ex.Data[AgentRefusalDataKey] is true; + + /// + /// The text of : the unexpected root, plus the Agent's own error text for an + /// MTConnectError document. + /// + /// The document's root element. + /// The root the caller expected. + /// How to describe the document. + /// The message. public static string DescribeUnexpectedRoot(XElement root, string expectedRootName, string subject) { var message = diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientBrowseSession.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientBrowseSession.cs index 42f10018..9224a80a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientBrowseSession.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientBrowseSession.cs @@ -17,6 +17,7 @@ internal sealed class OpcUaClientBrowseSession : IBrowseSession private readonly NodeId _rootNodeId; private readonly SemaphoreSlim _gate = new(1, 1); private volatile bool _disposed; + private readonly IDisposable? _ownedCredential; /// /// Construct a browse session bound to an already-connected . @@ -25,8 +26,10 @@ internal sealed class OpcUaClientBrowseSession : IBrowseSession /// Namespace snapshot taken at connect time; used to render outbound /// NodeIds in the server-stable nsu=… form. /// The node under which browses one level. - internal OpcUaClientBrowseSession(ISession session, NamespaceMap nsMap, NodeId rootNodeId) + /// A credential (the user certificate) this session owns and disposes with itself. + internal OpcUaClientBrowseSession(ISession session, NamespaceMap nsMap, NodeId rootNodeId, IDisposable? ownedCredential = null) { + _ownedCredential = ownedCredential; _session = session; _nsMap = nsMap; _rootNodeId = rootNodeId; @@ -54,10 +57,64 @@ internal sealed class OpcUaClientBrowseSession : IBrowseSession } /// - public Task> AttributesAsync(string nodeId, CancellationToken cancellationToken) + // R06:O-5 — reads the node's NodeClass / DisplayName / DataType / ValueRank / UserAccessLevel in one request and + // maps them through the same UpstreamAttributeMapping the runtime driver's discovery registers with, so the + // side-panel shows what a browse-commit would author. An Object (folder) has no value attributes → empty. + // Throws ArgumentException when nodeId cannot be resolved against the live session's namespace table. + public async Task> AttributesAsync(string nodeId, CancellationToken cancellationToken) { + ObjectDisposedException.ThrowIf(_disposed, this); + if (!NamespaceMap.TryResolve(_session, nodeId, out var resolved)) + throw new ArgumentException( + $"Cannot resolve NodeId '{nodeId}' against the live session.", nameof(nodeId)); + + uint[] attributeIds = + [ + Attributes.NodeClass, Attributes.DisplayName, Attributes.DataType, Attributes.ValueRank, + Attributes.UserAccessLevel, + ]; + var toRead = new ReadValueIdCollection(attributeIds.Length); + foreach (var id in attributeIds) toRead.Add(new ReadValueId { NodeId = resolved, AttributeId = id }); + + DataValueCollection values; + await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + var resp = await _session.ReadAsync( + requestHeader: null, + maxAge: 0, + timestampsToReturn: TimestampsToReturn.Neither, + nodesToRead: toRead, + ct: cancellationToken).ConfigureAwait(false); + values = resp?.Results ?? []; + } + finally + { + _gate.Release(); + } LastUsedUtc = DateTime.UtcNow; - return Task.FromResult>(Array.Empty()); + + DataValue? At(int i) => i < values.Count && StatusCode.IsGood(values[i].StatusCode) + ? values[i] : null; + + var nodeClass = At(0)?.Value is int nc ? (NodeClass)nc : NodeClass.Unspecified; + if (nodeClass != NodeClass.Variable) return Array.Empty(); + + var name = (At(1)?.Value as LocalizedText)?.Text ?? resolved.ToString() ?? nodeId; + var dataType = At(2)?.Value is NodeId dt + ? UpstreamAttributeMapping.MapDataType(dt) + : ZB.MOM.WW.OtOpcUa.Core.Abstractions.DriverDataType.String; + var valueRank = At(3)?.Value is int vr ? vr : ValueRanks.Scalar; + var access = At(4)?.Value is byte ab ? ab : (byte)0; + + return + [ + new AttributeInfo( + Name: name, + DriverDataType: dataType.ToString(), + IsArray: UpstreamAttributeMapping.IsArray(valueRank), + SecurityClass: UpstreamAttributeMapping.MapAccessLevel(access).ToString()), + ]; } /// Issue a single-level Browse (plus continuation-point follow-ups) under the @@ -188,5 +245,8 @@ internal sealed class OpcUaClientBrowseSession : IBrowseSession try { _gate.Dispose(); } catch { /* best-effort */ } + + try { _ownedCredential?.Dispose(); } + catch { /* best-effort */ } } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientDriverBrowser.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientDriverBrowser.cs index 6573e87d..da1411b1 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientDriverBrowser.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser/OpcUaClientDriverBrowser.cs @@ -7,7 +7,8 @@ using Opc.Ua.Client; using Opc.Ua.Configuration; using ZB.MOM.WW.OtOpcUa.Commons.Browsing; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; -using ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient; // OpcUaClientDriverOptions + NamespaceMap +using ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient; // OpcUaClientDriverOptions + NamespaceMap + OpcUaUserIdentityFactory +using ZB.MOM.WW.Secrets.Abstractions; namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Browser; @@ -33,12 +34,19 @@ public sealed class OpcUaClientDriverBrowser : IDriverBrowser }; private readonly ILogger _logger; + private readonly ISecretResolver? _secretResolver; /// Creates a new browser. Logger defaults to NullLogger when not supplied. /// Optional logger; defaults to . - public OpcUaClientDriverBrowser(ILogger? logger = null) + /// + /// The host's shared secret resolver (DI-injected where one is registered), used to resolve + /// secret:-prefixed Password / UserCertificatePassword exactly as the runtime driver does. + /// Null ⇒ a secret: ref fails closed (it is never sent as the literal). + /// + public OpcUaClientDriverBrowser(ILogger? logger = null, ISecretResolver? secretResolver = null) { _logger = logger ?? NullLogger.Instance; + _secretResolver = secretResolver; } /// @@ -58,58 +66,69 @@ public sealed class OpcUaClientDriverBrowser : IDriverBrowser if (string.IsNullOrWhiteSpace(endpoint)) throw new InvalidOperationException("OpcUaClient browser requires EndpointUrl or EndpointUrls[0]."); - if (opts.AuthType == OpcUaAuthType.Certificate) - throw new InvalidOperationException( - "Browser does not support OpcUaAuthType.Certificate in v1; use Anonymous or Username."); - - if (opts.AutoAcceptCertificates) - _logger.LogWarning( - "AdminUI browse session opens against {Endpoint} with form's AutoAcceptCertificates=true — " + - "browse uses its own cert store and does NOT auto-accept; trust the cert via the runtime " + - "driver's PKI store instead.", - endpoint); - - var appConfig = await BuildBrowseAppConfigurationAsync(cancellationToken).ConfigureAwait(false); - var identity = BuildBrowseUserIdentity(opts); - - var perEndpointBudget = TimeSpan.FromSeconds( - Math.Clamp(opts.PerEndpointConnectTimeout.TotalSeconds, 5, 30)); - using var connectCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - connectCts.CancelAfter(perEndpointBudget); - - var endpointDesc = await SelectEndpointAsync( - appConfig, endpoint, opts.SecurityPolicy, opts.SecurityMode, connectCts.Token).ConfigureAwait(false); - var endpointCfg = EndpointConfiguration.Create(appConfig); - endpointCfg.OperationTimeout = (int)opts.Timeout.TotalMilliseconds; - var configuredEndpoint = new ConfiguredEndpoint(null, endpointDesc, endpointCfg); - - var session = await new DefaultSessionFactory(telemetry: null!).CreateAsync( - appConfig, - configuredEndpoint, - updateBeforeConnect: false, - sessionName: "OtOpcUa AdminUI Browse", - (uint)opts.SessionTimeout.TotalMilliseconds, - identity, - preferredLocales: null, - connectCts.Token).ConfigureAwait(false); + // R06:O-5 — Certificate auth is supported: the identity is built (and any secret: credential resolved) by the + // same factory the runtime driver uses, BEFORE any network I/O, so a bad certificate path / password fails fast + // with its own message. + var (identity, ownedCertificate) = await BuildBrowseCredentialAsync(opts, _secretResolver, cancellationToken) + .ConfigureAwait(false); try { - var nsMap = NamespaceMap.FromSession(session); - var rootNodeId = string.IsNullOrEmpty(opts.BrowseRoot) - ? ObjectIds.ObjectsFolder - : NodeId.Parse(session.MessageContext, opts.BrowseRoot); + if (opts.AutoAcceptCertificates) + _logger.LogWarning( + "AdminUI browse session opens against {Endpoint} with form's AutoAcceptCertificates=true — " + + "browse uses its own cert store and does NOT auto-accept; trust the cert via the runtime " + + "driver's PKI store instead.", + endpoint); - _logger.LogInformation( - "AdminUI OPC UA browse session opened against {Endpoint} (policy {Policy}, mode {Mode})", - endpoint, opts.SecurityPolicy, opts.SecurityMode); + var appConfig = await BuildBrowseAppConfigurationAsync(cancellationToken).ConfigureAwait(false); - return new OpcUaClientBrowseSession(session, nsMap, rootNodeId); + var perEndpointBudget = TimeSpan.FromSeconds( + Math.Clamp(opts.PerEndpointConnectTimeout.TotalSeconds, 5, 30)); + using var connectCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + connectCts.CancelAfter(perEndpointBudget); + + var endpointDesc = await SelectEndpointAsync( + appConfig, endpoint, opts.SecurityPolicy, opts.SecurityMode, connectCts.Token).ConfigureAwait(false); + var endpointCfg = EndpointConfiguration.Create(appConfig); + endpointCfg.OperationTimeout = (int)opts.Timeout.TotalMilliseconds; + var configuredEndpoint = new ConfiguredEndpoint(null, endpointDesc, endpointCfg); + + var session = await new DefaultSessionFactory(telemetry: null!).CreateAsync( + appConfig, + configuredEndpoint, + updateBeforeConnect: false, + sessionName: "OtOpcUa AdminUI Browse", + (uint)opts.SessionTimeout.TotalMilliseconds, + identity, + preferredLocales: null, + connectCts.Token).ConfigureAwait(false); + + try + { + var nsMap = NamespaceMap.FromSession(session); + var rootNodeId = string.IsNullOrEmpty(opts.BrowseRoot) + ? ObjectIds.ObjectsFolder + : NodeId.Parse(session.MessageContext, opts.BrowseRoot); + + _logger.LogInformation( + "AdminUI OPC UA browse session opened against {Endpoint} (policy {Policy}, mode {Mode})", + endpoint, opts.SecurityPolicy, opts.SecurityMode); + + return new OpcUaClientBrowseSession(session, nsMap, rootNodeId, ownedCertificate); + } + catch + { + try { if (session is Session s) await s.CloseAsync().ConfigureAwait(false); } catch { /* best-effort */ } + try { session.Dispose(); } catch { /* best-effort */ } + throw; + } } catch { - try { if (session is Session s) await s.CloseAsync().ConfigureAwait(false); } catch { /* best-effort */ } - try { session.Dispose(); } catch { /* best-effort */ } + // Review m9: the certificate (and its private-key handle) is owned by the browse session once handed + // over; any failure before that releases it here instead of leaving it to the GC. + ownedCertificate?.Dispose(); throw; } } @@ -175,16 +194,39 @@ public sealed class OpcUaClientDriverBrowser : IDriverBrowser return config; } - /// Build the OPC UA user identity from the form's auth fields. + /// + /// Build the OPC UA user identity from the form's auth fields — Anonymous, Username or Certificate — through + /// the factory the runtime driver uses, after resolving any secret:-prefixed credential (fail-closed). + /// /// Driver options carrying the form's auth fields. - private static UserIdentity BuildBrowseUserIdentity(OpcUaClientDriverOptions opts) => opts.AuthType switch + /// The secret store, or null when none is available. + /// Cancellation token for the secret resolution. + /// The identity the browse session activates with. + internal static async Task BuildBrowseUserIdentityAsync( + OpcUaClientDriverOptions opts, ISecretResolver? secretResolver, CancellationToken ct) => + (await BuildBrowseCredentialAsync(opts, secretResolver, ct).ConfigureAwait(false)).Identity; + + /// + /// As , also handing back the user certificate a Certificate-auth + /// identity holds, so the browse session that uses it can dispose it (review m9: the browse opens one per picker + /// session, and an undisposed X509Certificate2 pins a private-key handle until GC). + /// + /// Driver options carrying the form's auth fields. + /// The secret store, or null when none is available. + /// Cancellation token for the secret resolution. + /// The identity, and the certificate it owns (null for Anonymous / Username). + internal static async Task<(UserIdentity Identity, System.Security.Cryptography.X509Certificates.X509Certificate2? Certificate)> + BuildBrowseCredentialAsync(OpcUaClientDriverOptions opts, ISecretResolver? secretResolver, CancellationToken ct) { - OpcUaAuthType.Anonymous => new UserIdentity(new AnonymousIdentityToken()), - OpcUaAuthType.Username => new UserIdentity( - opts.Username ?? string.Empty, - System.Text.Encoding.UTF8.GetBytes(opts.Password ?? string.Empty)), - _ => new UserIdentity(new AnonymousIdentityToken()), - }; + var resolved = await OpcUaUserIdentityFactory.ResolveSecretRefsAsync(opts, secretResolver, ct).ConfigureAwait(false); + if (resolved.AuthType != OpcUaAuthType.Certificate) + { + return (OpcUaUserIdentityFactory.Build(resolved), null); + } + + var certificate = OpcUaUserIdentityFactory.LoadCertificate(resolved); + return (new UserIdentity(certificate), certificate); + } /// Select the endpoint matching the requested SecurityPolicy + SecurityMode pair. /// Application configuration used by the discovery client. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/OpcUaUserIdentityFactory.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/OpcUaUserIdentityFactory.cs new file mode 100644 index 00000000..29510067 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/OpcUaUserIdentityFactory.cs @@ -0,0 +1,126 @@ +using Opc.Ua; +using ZB.MOM.WW.Secrets.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient; + +/// +/// The ONE place the OpcUaClient user identity is built from — shared by +/// the runtime driver's session open and the AdminUI live browse, so a certificate-authenticated upstream is +/// browsable with exactly the credentials the deployed driver will present. Includes the secret: +/// resolution of the two credential fields (fail-closed). +/// +public static class OpcUaUserIdentityFactory +{ + private const string SecretPrefix = "secret:"; + + /// + /// Build the user-identity token from options whose credential fields are already resolved plaintext + /// (see ). + /// + /// Driver options containing authentication configuration. + /// The constructed for session activation. + public static UserIdentity Build(OpcUaClientDriverOptions options) => + options.AuthType switch + { + OpcUaAuthType.Anonymous => new UserIdentity(new AnonymousIdentityToken()), + OpcUaAuthType.Username => new UserIdentity( + options.Username ?? string.Empty, + System.Text.Encoding.UTF8.GetBytes(options.Password ?? string.Empty)), + OpcUaAuthType.Certificate => FromCertificate(options), + _ => new UserIdentity(new AnonymousIdentityToken()), + }; + + /// + /// Build a carrying a client user-authentication certificate loaded from + /// . Fails fast if the path is missing, the file + /// doesn't exist, or the certificate lacks a private key (required to sign the user-token challenge). + /// + /// Driver options containing certificate configuration. + /// The certificate-backed . + public static UserIdentity FromCertificate(OpcUaClientDriverOptions options) => new(LoadCertificate(options)); + + /// + /// Loads (and validates) the user-authentication certificate wraps. The caller owns + /// the returned certificate — dispose it when the session that presented it is gone. + /// + /// Driver options containing certificate configuration. + /// The certificate, with its private key. + public static System.Security.Cryptography.X509Certificates.X509Certificate2 LoadCertificate(OpcUaClientDriverOptions options) + { + if (string.IsNullOrWhiteSpace(options.UserCertificatePath)) + throw new InvalidOperationException( + "OpcUaAuthType.Certificate requires OpcUaClientDriverOptions.UserCertificatePath to be set."); + if (!File.Exists(options.UserCertificatePath)) + throw new FileNotFoundException( + $"User certificate not found at '{options.UserCertificatePath}'.", + options.UserCertificatePath); + + // X509CertificateLoader (new in .NET 9) is the only non-obsolete way to load a PFX + // since the legacy X509Certificate2 ctors are marked obsolete on net10. + var cert = System.Security.Cryptography.X509Certificates.X509CertificateLoader + .LoadPkcs12FromFile(options.UserCertificatePath, options.UserCertificatePassword); + + if (!cert.HasPrivateKey) + { + cert.Dispose(); + throw new InvalidOperationException( + $"User certificate at '{options.UserCertificatePath}' has no private key — " + + "the private key is required to sign the OPC UA user-token challenge at session activation."); + } + + return cert; + } + + /// + /// Return a copy of with any secret:-prefixed + /// / + /// resolved through . Non-secret / null / empty fields pass through unchanged (the + /// reference-equal original is returned when nothing changed). Fail-closed: an absent secret, or a + /// secret: ref with no resolver available, throws rather than sending the literal as a credential. + /// + /// The raw options (credential fields may carry secret: refs). + /// The shared secret resolver, or null when none is available. + /// Cancellation token. + /// An options copy whose credential fields carry resolved plaintext. + public static async Task ResolveSecretRefsAsync( + OpcUaClientDriverOptions options, ISecretResolver? resolver, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(options); + + var password = await ResolveFieldAsync(options.Password, nameof(options.Password), resolver, ct) + .ConfigureAwait(false); + var certPassword = await ResolveFieldAsync( + options.UserCertificatePassword, nameof(options.UserCertificatePassword), resolver, ct) + .ConfigureAwait(false); + + if (ReferenceEquals(password, options.Password) + && ReferenceEquals(certPassword, options.UserCertificatePassword)) + { + return options; + } + + return options with { Password = password, UserCertificatePassword = certPassword }; + } + + private static async Task ResolveFieldAsync( + string? value, string fieldName, ISecretResolver? resolver, CancellationToken ct) + { + if (string.IsNullOrEmpty(value) + || !value.StartsWith(SecretPrefix, StringComparison.OrdinalIgnoreCase)) + { + return value; + } + + var name = value[SecretPrefix.Length..]; + if (resolver is null) + throw new InvalidOperationException( + $"OpcUaClientDriverOptions.{fieldName}='{value}' references secret '{name}', but no secret store is " + + "available here (fail-closed)."); + var resolved = await resolver.GetAsync(new SecretName(name), ct).ConfigureAwait(false); + return !string.IsNullOrEmpty(resolved) + ? resolved + : throw new InvalidOperationException( + $"OpcUaClientDriverOptions.{fieldName}='{value}' resolves secret '{name}', but it is " + + "absent from the store (fail-closed)."); + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/UpstreamAttributeMapping.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/UpstreamAttributeMapping.cs new file mode 100644 index 00000000..046fa2e1 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/UpstreamAttributeMapping.cs @@ -0,0 +1,65 @@ +using Opc.Ua; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient; + +/// +/// The ONE mapping from an upstream variable's OPC UA attributes (DataType, ValueRank, AccessLevel) to the +/// driver-agnostic shape — shared by the runtime driver's discovery enrichment and the AdminUI live browse +/// side-panel, so the picker shows exactly what a browse-commit / discovery would register. +/// +public static class UpstreamAttributeMapping +{ + /// + /// Map an upstream OPC UA built-in DataType NodeId (via DataTypeIds.*) to a + /// . Unknown / custom types fall through to + /// which is the safest passthrough for + /// Variant-wrapped structs + enums + extension objects. + /// + /// The OPC UA data type NodeId to map. + /// The mapped . + public static DriverDataType MapDataType(NodeId dataType) + { + if (dataType == DataTypeIds.Boolean) return DriverDataType.Boolean; + // SByte (signed 8-bit) shares Int16 — DriverDataType has no narrower signed type. + // Byte (unsigned 8-bit) belongs in the unsigned family → UInt16, not Int16 + // (mapping an unsigned 0-255 type onto Int16 misrepresents type metadata and + // confuses range/validation logic keyed off DriverDataType). + if (dataType == DataTypeIds.SByte || dataType == DataTypeIds.Int16) return DriverDataType.Int16; + if (dataType == DataTypeIds.Byte || dataType == DataTypeIds.UInt16) return DriverDataType.UInt16; + if (dataType == DataTypeIds.Int32) return DriverDataType.Int32; + if (dataType == DataTypeIds.UInt32) return DriverDataType.UInt32; + if (dataType == DataTypeIds.Int64) return DriverDataType.Int64; + if (dataType == DataTypeIds.UInt64) return DriverDataType.UInt64; + if (dataType == DataTypeIds.Float) return DriverDataType.Float32; + if (dataType == DataTypeIds.Double) return DriverDataType.Float64; + if (dataType == DataTypeIds.String) return DriverDataType.String; + if (dataType == DataTypeIds.DateTime || dataType == DataTypeIds.UtcTime) + return DriverDataType.DateTime; + return DriverDataType.String; + } + + /// + /// Map an OPC UA AccessLevel/UserAccessLevel attribute value (AccessLevels bitmask) to a + /// : CurrentWrite-capable variables surface as + /// ; read-only as . + /// + /// The OPC UA access level bitmask. + /// The mapped security classification. + public static SecurityClassification MapAccessLevel(byte accessLevel) + { + const byte CurrentWrite = 2; // AccessLevels.CurrentWrite = 0x02 + return (accessLevel & CurrentWrite) != 0 + ? SecurityClassification.Operate + : SecurityClassification.ViewOnly; + } + + /// + /// OPC UA Part 3 ValueRank: -3 ScalarOrOneDimension, -2 Any, -1 Scalar, 0 OneOrMoreDimensions, 1 OneDimension, + /// >1 N dimensions. Anything >= 0 is an array; -3 / -2 are conservatively scalar until the upstream variable + /// carries a concrete dimensioned rank. + /// + /// The ValueRank attribute. + /// Whether the variable is exposed as an array. + public static bool IsArray(int valueRank) => valueRank >= 0; +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session.csproj b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session.csproj index 3ecc9663..41166a40 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session.csproj +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Session.csproj @@ -11,8 +11,9 @@ + +