From b548cc46e681698d2d22528bcfcb66a38cdc1081 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:39:58 -0400 Subject: [PATCH] perf(mtconnect): read through the live pump index; instance-id check on the /current fallback (R06:T-4, T-2) - T-4: every ReadAsync batch fetched and parsed the whole device /current and built a fresh index even while the /sample pump held a live one. A read is now answered from the pump's index (read only - the pump stays its single writer) while the current-generation pump is running, has delivered since it started and is not degraded; /current stays the fallback, and its per-batch index is built over a tag map frozen once per binding instead of per call. A served read clears the read-path degradation a failed /current left. - T-2: the /current fallback checks the answer's instanceId; a change (an Agent restart seen by a browse-only driver) drops the cached /probe model and raises rediscovery once per distinct id, only while the read's session is still installed. - The pump also acts on no verdict (Degrade / latch) once its session generation is gone, so an abandoned pump cannot write health over the new session (T-1 follow-through). Refs #743 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- docs/drivers/MTConnect.md | 11 +- .../MTConnectDriver.cs | 96 ++++++++++- .../MTConnectObservationIndex.cs | 29 +++- .../MTConnectReadPathTests.cs | 162 ++++++++++++++++++ 4 files changed, 293 insertions(+), 5 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectReadPathTests.cs diff --git a/docs/drivers/MTConnect.md b/docs/drivers/MTConnect.md index fe22a648..9e338bb0 100644 --- a/docs/drivers/MTConnect.md +++ b/docs/drivers/MTConnect.md @@ -87,7 +87,7 @@ 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) | @@ -113,6 +113,15 @@ 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. 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. + ## Minimum deployment ```jsonc 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 cbbdb266..0440c81c 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,13 @@ public sealed partial class MTConnectDriver /// Whether the /current read path is currently failing. See . private volatile bool _readDegraded; + /// + /// 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 @@ -713,14 +720,33 @@ 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) + && ReferenceEquals(Volatile.Read(ref _session)?.Client, client)) + { + AnnounceAgentRestart(current.InstanceId, options); + } + // 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]; @@ -752,6 +778,47 @@ 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. + /// + /// 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. + if (_readDegraded) + { + _readDegraded = false; + PublishHealthy(ReadHealth().LastSuccessfulRead ?? DateTime.UtcNow); + } + + return results; + } + // ---- IWritable: observation injection through the Agent's AllowPut surface ---- /// @@ -1857,6 +1924,9 @@ public sealed partial class MTConnectDriver _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, generation, cts.Token), @@ -1935,6 +2005,13 @@ public sealed partial class MTConnectDriver 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; + } + switch (outcome.Verdict) { case StreamVerdict.Cancelled: @@ -2007,8 +2084,11 @@ 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); + if (IsCurrentSessionGeneration(generation)) + { + _streamDegraded = true; + Degrade(ex); + } } finally { @@ -2020,6 +2100,10 @@ public sealed partial class MTConnectDriver // 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; + } } } @@ -2112,6 +2196,7 @@ public sealed partial class MTConnectDriver expected = chunk.NextSequence; _streamDegraded = false; + _pumpLive = true; PublishHealthy(DateTime.UtcNow); } @@ -2196,6 +2281,7 @@ public sealed partial class MTConnectDriver ApplyAndFanOut(current, baseline: true); _streamDegraded = false; + _pumpLive = true; PublishHealthy(DateTime.UtcNow); _logger.LogInformation( @@ -3692,6 +3778,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. /// 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/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectReadPathTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectReadPathTests.cs new file mode 100644 index 00000000..8c66a278 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectReadPathTests.cs @@ -0,0 +1,162 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests; + +/// +/// Review 2026-10-21 R06:T-4 and T-2 on the read path. +/// +/// T-4 — every ReadAsync batch fetched + parsed the whole device /current even while the +/// /sample pump held a live, healthy index. A read now answers from that index (READ only — the pump +/// stays its single writer) while the pump is live and not degraded, and keeps /current as the +/// fallback. +/// T-2 — Agent-restart detection lived only on the pump, and the read path discarded the +/// /current answer's instanceId: a browse-only driver kept a cached /probe model across an +/// Agent restart and never raised rediscovery. The fallback read now checks it. +/// +/// +public sealed class MTConnectReadPathTests +{ + private const long RestartedInstance = 1655000777L; + private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(30); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static MTConnectDriverOptions Opts() => new() + { + AgentUri = "http://fixture-agent:5000", + RequestTimeoutMs = 5000, + Tags = [new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64)], + }; + + /// T-4: with a live, healthy pump the read is answered from its index — no /current round trip. + [Fact] + public async Task A_read_with_a_live_pump_answers_from_the_index_without_a_current_call() + { + var client = CannedAgentClient.FromFixtures(); + var driver = new MTConnectDriver(Opts(), "mt-t4", _ => client); + await driver.InitializeAsync("{}", Ct); + try + { + await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + await client.PumpAsync(MTConnectPumpGenerationChunks.Pos(108, 113, "7.25")).WaitAsync(Watchdog, Ct); + var currentCalls = client.CurrentCallCount; + + var result = (await driver.ReadAsync(["dev1_pos"], Ct)).Single(); + + result.Value.ShouldBe(7.25d, "the read must serve the pump's newest value"); + client.CurrentCallCount.ShouldBe(currentCalls, "no /current while the pump's index is live"); + } + finally + { + await driver.ShutdownAsync(CancellationToken.None); + } + } + + /// Positive control for T-4: with no pump the read still issues exactly one /current. + [Fact] + public async Task A_read_with_no_pump_falls_back_to_one_current_call() + { + var client = CannedAgentClient.FromFixtures(); + var driver = new MTConnectDriver(Opts(), "mt-t4-fallback", _ => client); + await driver.InitializeAsync("{}", Ct); + try + { + var currentCalls = client.CurrentCallCount; + + await driver.ReadAsync(["dev1_pos"], Ct); + + client.CurrentCallCount.ShouldBe(currentCalls + 1); + } + finally + { + await driver.ShutdownAsync(CancellationToken.None); + } + } + + /// T-4: a pump whose stream is failing does not serve reads — they fall back to /current. + [Fact] + public async Task A_read_with_a_degraded_pump_falls_back_to_current() + { + var client = CannedAgentClient.FromFixtures(); + // Every /sample fails: the pump stays alive (reconnecting under backoff) but its stream is degraded. + client.SampleFailure = new IOException("connection refused"); + var options = new MTConnectDriverOptions + { + AgentUri = "http://fixture-agent:5000", + RequestTimeoutMs = 5000, + Tags = [new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64)], + Reconnect = new MTConnectReconnectOptions { MinBackoffMs = 1000, MaxBackoffMs = 1000 }, + }; + var driver = new MTConnectDriver(options, "mt-t4-degraded", _ => client); + await driver.InitializeAsync("{}", Ct); + try + { + await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + await client.WaitForSampleCallsAsync(1).WaitAsync(Watchdog, Ct); + var deadline = DateTime.UtcNow + Watchdog; + while (driver.GetHealth().State != DriverState.Degraded && DateTime.UtcNow < deadline) + await Task.Delay(20, Ct); + driver.GetHealth().State.ShouldBe(DriverState.Degraded); + var currentCalls = client.CurrentCallCount; + + await driver.ReadAsync(["dev1_pos"], Ct); + + client.CurrentCallCount.ShouldBe(currentCalls + 1, "a degraded stream's index is not trusted for reads"); + } + finally + { + await driver.ShutdownAsync(CancellationToken.None); + } + } + + /// T-2: a browse-only driver's read sees the Agent restart — once — and drops the stale model. + [Fact] + public async Task A_read_that_sees_a_new_instance_id_raises_rediscovery_once_and_drops_the_probe_model() + { + var client = CannedAgentClient.FromFixtures(); + var driver = new MTConnectDriver(Opts(), "mt-t2", _ => client); + await driver.InitializeAsync("{}", Ct); + var raised = 0; + driver.OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref raised); + try + { + driver.CachedProbeModel.ShouldNotBeNull("positive control: initialize cached the model"); + client.Current = new MTConnectStreamsResult(RestartedInstance, 5, 1, client.Current.Observations); + + await driver.ReadAsync(["dev1_pos"], Ct); + await driver.ReadAsync(["dev1_pos"], Ct); + + raised.ShouldBe(1, "once per distinct instanceId, not once per read"); + driver.CachedProbeModel.ShouldBeNull("the model describes a process that no longer exists"); + } + finally + { + await driver.ShutdownAsync(CancellationToken.None); + } + } + + /// T-2 control: reads against the same Agent process never raise rediscovery. + [Fact] + public async Task Reads_against_the_same_instance_raise_nothing() + { + var client = CannedAgentClient.FromFixtures(); + var driver = new MTConnectDriver(Opts(), "mt-t2-same", _ => client); + await driver.InitializeAsync("{}", Ct); + var raised = 0; + driver.OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref raised); + try + { + await driver.ReadAsync(["dev1_pos"], Ct); + await driver.ReadAsync(["dev1_pos"], Ct); + + raised.ShouldBe(0); + driver.CachedProbeModel.ShouldNotBeNull(); + } + finally + { + await driver.ShutdownAsync(CancellationToken.None); + } + } +}