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
This commit is contained in:
Joseph Doherty
2026-10-22 15:39:58 -04:00
parent 77bfadbdda
commit b548cc46e6
4 changed files with 293 additions and 5 deletions
@@ -369,6 +369,13 @@ public sealed partial class MTConnectDriver
/// <summary>Whether the <c>/current</c> read path is currently failing. See <see cref="PublishHealthy"/>.</summary>
private volatile bool _readDegraded;
/// <summary>
/// Whether the current-generation <c>/sample</c> pump has delivered since it started (a chunk, a heartbeat or
/// a re-baseline) — the precondition, with <see cref="_streamDegraded"/> clear, for <see cref="ReadAsync"/>
/// to answer from the pump's index instead of a <c>/current</c> (R06:T-4).
/// </summary>
private volatile bool _pumpLive;
/// <summary>
/// Consecutive <c>/sample</c> reconnect attempts that have <b>not</b> yet been vindicated by
/// a delivered chunk — the input to <see cref="BackoffFor"/>. 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
}
}
/// <summary>
/// Answers a read batch from the live pump's index, or returns <see langword="null"/> 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.
/// </summary>
/// <param name="fullReferences">The batch.</param>
/// <param name="binding">The tag table the batch is resolved against.</param>
/// <returns>One snapshot per reference in order, or <see langword="null"/> to fall back to <c>/current</c>.</returns>
private DataValueSnapshot[]? TryReadFromLivePump(IReadOnlyList<string> 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 ----
/// <inheritdoc />
@@ -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<MTConnectTagDefinition> IndexTags,
FrozenDictionary<string, WriteTarget> WriteTargetsByDataItemId)
{
/// <summary><see cref="IndexTags"/> frozen once per binding, for the per-read index (R06:T-4).</summary>
public FrozenDictionary<string, MTConnectTagDefinition> IndexTagMap { get; } =
MTConnectObservationIndex.FreezeTags(IndexTags);
/// <summary>
/// Resolves a write reference to the DataItem id and the Agent address it writes through.
/// </summary>
@@ -151,7 +151,34 @@ internal sealed class MTConnectObservationIndex
TimeProvider? timeProvider = null)
{
_timeProvider = timeProvider ?? TimeProvider.System;
_tags = FreezeTags(tags);
}
/// <summary>Builds an index over an already-frozen tag map (see <see cref="FreezeTags"/>).</summary>
private MTConnectObservationIndex(FrozenDictionary<string, MTConnectTagDefinition> frozenTags, TimeProvider? timeProvider)
{
_timeProvider = timeProvider ?? TimeProvider.System;
_tags = frozenTags;
}
/// <summary>
/// R06:T-4 — an index over a tag map frozen ONCE (per binding) rather than per construction: the
/// <c>/current</c> 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.
/// </summary>
/// <param name="frozenTags">A map produced by <see cref="FreezeTags"/>.</param>
/// <returns>A fresh, empty index coercing against <paramref name="frozenTags"/>.</returns>
internal static MTConnectObservationIndex OverFrozenTags(FrozenDictionary<string, MTConnectTagDefinition> frozenTags) =>
new(frozenTags, null);
/// <summary>
/// The authored tags keyed by <see cref="MTConnectTagDefinition.FullName"/>, last definition winning on a
/// duplicate (see the constructor), blank / null entries skipped.
/// </summary>
/// <param name="tags">The authored tags.</param>
/// <returns>The frozen map an index coerces against.</returns>
internal static FrozenDictionary<string, MTConnectTagDefinition> FreezeTags(IEnumerable<MTConnectTagDefinition>? tags)
{
var map = new Dictionary<string, MTConnectTagDefinition>(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);
}
/// <summary>The number of distinct dataItemIds currently indexed.</summary>