dohertj2/lmxopcua
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ablegacy): any answer restores Healthy; writes feed health and the backoff

Browse Source 
Review r0 I2 + M1/M2/M4/M7/M10.
- I2: one health writer (ApplyHealth) for reads AND writes, worst wins per
  call: a comms fault => Degraded; otherwise any answer (Good value, device
  refusal, Good or refused write) => Healthy; LastSuccessfulRead advances on
  any answer. A refusal-only read used to leave a recovered driver Degraded,
  and writes never touched health at all.
- M10: a no-reply write STATUS now degrades like the exception path did; a
  Good/refused write records an answer on the device's backoff, a no-reply
  one a fault (bit RMW included — it now reports its raw status).
- M2: a write inside the device's backoff window fails fast with
  BadCommunicationError even when its handle is cached.
- M1: ShutdownAsync/Dispose write their final Unknown only while the epoch
  they retired is still current (a newer initialise owns health otherwise).
- M4: an indirect bit write releases the previous index value's parent-word
  handle (the per-word RMW gate is deliberately kept: removing a gate another
  writer may hold would let two RMWs of one word race).
- M7: backoff decisions read an injectable Clock; the isolation tests drop
  their stopwatch asserts and cross the window on a manual clock.

Refs #738
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 15:16:49 -04:00
parent 8a80dcc787
commit 17ae2eea0b
3 changed files with 360 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs
+134 -54
View File
@@ -177,7 +177,7 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
// allows it, and a reconnect retry can overlap a running attempt). Release the previous device states' probe loops + runtimes BEFORE the
// new table is published, or they are orphaned beyond the reach of ShutdownAsync. Deliberately NOT
// ShutdownAsync: that also drops _poll and every subscription, which are owned by the shutdown path (the actor re-subscribes after every successful init).
await ReleaseDeviceStatesAsync(RetireTables(bumpEpoch: false), "re-initialise").ConfigureAwait(false);
await ReleaseDeviceStatesAsync(RetireTables(bumpEpoch: false).States, "re-initialise").ConfigureAwait(false);
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Initializing, null, null));
_logger.LogInformation(
"AbLegacy driver {DriverInstanceId} initializing: transport={Transport} tagFactory={TagFactory}",
@@ -328,8 +328,9 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
/// is clearing.
/// </summary>
/// <param name="bumpEpoch">True for a teardown: supersedes any in-flight initialise.</param>
/// <returns>The device states that were published.</returns>
private List<DeviceState> RetireTables(bool bumpEpoch)
/// <returns>The device states that were published, and the epoch as of the retirement (a teardown writes its final
/// health only while that epoch is still current — review 07 r0 M1).</returns>
private (List<DeviceState> States, long Epoch) RetireTables(bool bumpEpoch)
{
lock (_lifecycleLock)
{
@@ -340,7 +341,7 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
}
var retired = _devices.Values.ToList();
_devices = new Dictionary<string, DeviceState>(StringComparer.OrdinalIgnoreCase);
return retired;
return (retired, _lifecycleEpoch);
}
}
@@ -401,15 +402,19 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
public async Task ShutdownAsync(CancellationToken cancellationToken)
{
// Unpublish FIRST, so an in-flight initialise is superseded and new reads resolve nothing.
var retired = RetireTables(bumpEpoch: true);
var (retired, teardownEpoch) = RetireTables(bumpEpoch: true);
await _poll.DisposeAsync().ConfigureAwait(false);
// Cancel the probe loops, JOIN them (bounded), then dispose the CTSs and runtimes.
// The loop used to be fire-and-forget, so one parked in a probe-runtime create/read could
// outlive shutdown and keep a libplctag handle alive past retirement.
await ReleaseDeviceStatesAsync(retired, "shutdown").ConfigureAwait(false);
_resolver.Clear(); // drop transient equipment-tag parses so a config change re-parses
// A newer InitializeAsync that ran while this teardown was still joining owns health now.
lock (_lifecycleLock)
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
{
if (teardownEpoch == _lifecycleEpoch)
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
}
}
/// <inheritdoc />
@@ -445,6 +450,13 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
/// </summary>
internal const int FaultsToMarkDeviceDown = 2;
/// <summary>
/// The clock every per-device backoff decision reads (window open? when does it end?). Wall-clock UTC in
/// production; a test substitutes a manual clock so the window can be crossed deterministically rather than by
/// sleeping (review 07 r0 M7).
/// </summary>
internal Func<DateTime> Clock { get; set; } = () => DateTime.UtcNow;
/// <summary>One resolved reference queued for its device's pass.</summary>
private sealed record ReadItem(int Index, string Reference, AbLegacyTagDefinition Def, AbLegacyAddress Parsed);
@@ -552,19 +564,30 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
anyAnswered |= o.AnyAnswered;
fault ??= o.FaultMessage;
}
if (fault is null && !anyGood && !anyAnswered) return;
ApplyHealth(epoch, fault, anyGood, anyAnswered, now);
}
/// <summary>
/// The single health write for a read or write call (worst wins): any comms fault ⇒ Degraded; otherwise any
/// answer — a Good value OR a device refusal, both prove the device replied — ⇒ Healthy (review 07 r0 I2: a
/// refusal-only pass used to leave a recovered driver Degraded). <c>LastSuccessfulRead</c> advances whenever
/// any device answered. A call that reached no wire changes nothing; a Faulted (config-level) state is kept.
/// Epoch-guarded so a late call cannot rewrite the health of a teardown or newer initialise.
/// </summary>
private void ApplyHealth(long epoch, string? fault, bool anyGood, bool anyAnswered, DateTime now)
{
if (fault is null && !anyGood && !anyAnswered) return;
lock (_lifecycleLock)
{
if (epoch != _lifecycleEpoch) return;
var current = _health;
var lastRead = anyAnswered ? now : current.LastSuccessfulRead;
if (fault is not null)
_health = new DriverHealth(DriverState.Degraded, lastRead, fault);
else if (anyGood)
_health = new DriverHealth(DriverState.Healthy, lastRead, null);
else
if (current.State == DriverState.Faulted)
_health = current with { LastSuccessfulRead = lastRead };
else if (fault is not null)
_health = new DriverHealth(DriverState.Degraded, lastRead, fault);
else
_health = new DriverHealth(DriverState.Healthy, lastRead, null);
}
}
@@ -577,7 +600,7 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
while (work.TryDequeue(out var unit))
{
var members = unit.Block?.Members ?? [unit.Single!];
if (device.IsThrottled(DateTime.UtcNow))
if (device.IsThrottled(Clock()))
{
foreach (var m in members)
results[m.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadCommunicationError, null, now);
@@ -613,7 +636,7 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
break;
case OpVerdict.Fault:
outcome.FaultMessage ??= message;
device.RecordCommsFault(DateTime.UtcNow);
device.RecordCommsFault(Clock());
break;
case OpVerdict.Throttled:
outcome.FaultMessage ??= message;
@@ -785,6 +808,9 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
var epoch = CurrentEpoch;
var results = new WriteResult[writes.Count];
var devices = _devices; // one snapshot of the published table for the whole call
// Writes feed health too (review 07 r0 I2): a Good or device-refused write proves the device answered; a
// no-reply failure is a comms fault. Collected per call and written once, worst wins.
var outcome = new DeviceReadOutcome();
for (var i = 0; i < writes.Count; i++)
{
@@ -817,6 +843,15 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
continue;
}
// Inside the device's backoff window nothing is dialled — the write fails fast exactly as a read
// does, instead of paying the full Timeout on a handle that is still cached (review 07 r0 M2).
if (device.IsThrottled(Clock()))
{
results[i] = new WriteResult(AbLegacyStatusMapper.BadCommunicationError);
outcome.FaultMessage ??= $"device {device.Options.HostAddress} is in its reconnect backoff window";
continue;
}
// Indirect: resolve the index first (non-atomic with the write - the index may change
// between the two PLC accesses), then write the concrete address through the normal path.
if (parsed.IsIndirect)
@@ -826,8 +861,10 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
{
results[i] = new WriteResult(res.Status);
if (res.TransportFault)
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead,
$"transport fault resolving indirect address for {w.FullReference}"));
{
device.RecordCommsFault(Clock());
outcome.FaultMessage ??= $"transport fault resolving indirect address for {w.FullReference}";
}
continue;
}
def = res.Def;
@@ -872,8 +909,10 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
// may reject it; we surface that real PCCC status rather than pre-rejecting at the driver.
else if (def.DataType == AbLegacyDataType.Bit && parsed.BitIndex is int bit)
{
results[i] = new WriteResult(
await WriteBitInWordAsync(device, parsed, bit, w.Value, cancellationToken).ConfigureAwait(false));
var (code, raw) = await WriteBitInWordAsync(device, def, parsed, bit, w.Value, cancellationToken)
.ConfigureAwait(false);
results[i] = new WriteResult(code);
RecordWriteStatus(device, def, raw, ref outcome);
continue;
}
@@ -896,36 +935,23 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
opLock.Release();
}
if (status == 0)
{
device.RecordAnswered();
results[i] = new WriteResult(AbLegacyStatusMapper.Good);
}
else if (IsDeviceRefusal((AbStatus)status))
{
// The device answered and refused this write deterministically: keep the handle, warn once.
device.RecordAnswered();
WarnRefusalOnce(device, def, "write", $"device status {(AbStatus)status}");
results[i] = new WriteResult(AbLegacyStatusMapper.MapStatus(status));
}
else
{
// Evict the stale handle so the next call re-creates it (mirrors AbCip).
EvictRuntime(device, def.Name);
device.RecordCommsFault(DateTime.UtcNow);
results[i] = new WriteResult(AbLegacyStatusMapper.MapStatus(status));
}
results[i] = new WriteResult(status == 0 ? AbLegacyStatusMapper.Good : AbLegacyStatusMapper.MapStatus(status));
// A no-reply status evicts the stale handle so the next call re-creates it (mirrors AbCip).
if (IsNoReply((AbStatus)status) && status != 0) EvictRuntime(device, def.Name);
RecordWriteStatus(device, def, status, ref outcome);
}
catch (OperationCanceledException) { throw; }
catch (ConnectionThrottledException)
catch (ConnectionThrottledException tex)
{
// The device is in its reconnect backoff window: fail fast, no wire, no new evidence.
// The device is in its reconnect backoff window: fail fast, no wire, no new evidence for the backoff.
results[i] = new WriteResult(AbLegacyStatusMapper.BadCommunicationError);
outcome.FaultMessage ??= tex.Message;
}
catch (AbTransportException ex) when (IsDeviceRefusal(ex.Status))
{
device.RecordAnswered();
WarnRefusalOnce(device, def, "write", ex.Message);
outcome.AnyAnswered = true;
WarnRefusalOnce(device, def, "write", ex.Message, AbLegacyStatusMapper.MapStatus(ex.Status));
results[i] = new WriteResult(AbLegacyStatusMapper.MapStatus(ex.Status));
}
catch (NotSupportedException nse)
@@ -953,15 +979,38 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
{
// Transport exception — evict so the next write/read creates a fresh handle (mirrors AbCip).
EvictRuntime(device, def.Name);
device.RecordCommsFault(DateTime.UtcNow);
device.RecordCommsFault(Clock());
results[i] = new WriteResult(AbLegacyStatusMapper.BadCommunicationError);
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message));
outcome.FaultMessage ??= ex.Message;
}
}
ApplyHealth(epoch, outcome.FaultMessage, anyGood: outcome.AnyAnswered, anyAnswered: outcome.AnyAnswered, DateTime.UtcNow);
return results;
}
/// <summary>
/// Classifies a completed write's raw status into the call's health evidence and the device's backoff:
/// Good or a device refusal = answered; a no-reply status = comms fault (it used to leave health untouched on
/// this path while the exception path degraded — review 07 r0 M10); Unsupported = neutral.
/// </summary>
private void RecordWriteStatus(DeviceState device, AbLegacyTagDefinition def, int status, ref DeviceReadOutcome outcome)
{
var s = (AbStatus)status;
if (status == 0 || IsDeviceRefusal(s))
{
device.RecordAnswered();
outcome.AnyAnswered = true;
if (status != 0)
WarnRefusalOnce(device, def, "write", $"device status {s}", AbLegacyStatusMapper.MapStatus(status));
}
else if (IsNoReply(s))
{
device.RecordCommsFault(Clock());
outcome.FaultMessage ??= $"transport status {status} writing {def.Name}";
}
}
// ---- ITagDiscovery ----
/// <inheritdoc />
@@ -1052,7 +1101,7 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
// The probe's verdict feeds the data path's backoff (review 07 P7): an answer resets the window so a
// recovered device is read again at once; a comms fault counts toward marking it down.
if (success || IsDeviceRefusal((AbStatus)status)) state.RecordAnswered();
else if ((AbStatus)status != AbStatus.Unsupported) state.RecordCommsFault(DateTime.UtcNow);
else if ((AbStatus)status != AbStatus.Unsupported) state.RecordCommsFault(Clock());
}
catch (OperationCanceledException) when (ct.IsCancellationRequested) { break; }
catch (Exception ex)
@@ -1061,7 +1110,7 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
probeRuntime = null;
state.ProbeInitialized = false;
if (ex is AbTransportException ate && IsDeviceRefusal(ate.Status)) state.RecordAnswered();
else if (!IsLocalRefusal(ex)) state.RecordCommsFault(DateTime.UtcNow);
else if (!IsLocalRefusal(ex)) state.RecordCommsFault(Clock());
}
TransitionDeviceState(state, success ? HostState.Running : HostState.Stopped);
@@ -1155,8 +1204,10 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
/// <see cref="SemaphoreSlim"/>. The parent word is masked to its native width before the
/// bit operation so a sign-extended decode never corrupts the high bits.
/// </summary>
private async Task<uint> WriteBitInWordAsync(
AbLegacyDriver.DeviceState device, AbLegacyAddress bitAddress, int bit, object? value, CancellationToken ct)
/// <returns>The OPC UA status the write is answered with, and the raw transport status (0 = Good) for health.</returns>
private async Task<(uint Code, int Raw)> WriteBitInWordAsync(
AbLegacyDriver.DeviceState device, AbLegacyTagDefinition def, AbLegacyAddress bitAddress, int bit, object? value,
CancellationToken ct)
{
// The parent word width follows the file letter: an L-file element is 32-bit, every
// other bit-addressable PCCC file (N/I/O/S/A) is a 16-bit word. bit is already
@@ -1169,6 +1220,23 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
var parentAddress = bitAddress with { BitIndex = null };
var parentName = parentAddress.ToLibplctagName();
// An indirect bit write (N7:[N7:0]/3) resolves to a different parent word per index value: release the
// previous resolution's parent handle so a sweeping index does not keep one per word (review 07 r0 M4). The
// per-word RMW gate is kept — removing a gate another writer may hold would let two RMWs of one word race.
var arrow = def.Name.IndexOf("=>", StringComparison.Ordinal);
if (arrow > 0)
{
var indirectName = def.Name[..arrow];
var previous = device.IndirectLastParent.TryGetValue(indirectName, out var p) ? p : null;
device.IndirectLastParent[indirectName] = parentName;
if (previous is not null && !string.Equals(previous, parentName, StringComparison.OrdinalIgnoreCase))
{
var oldGate = device.GetRmwLock(previous);
await oldGate.WaitAsync(ct).ConfigureAwait(false);
try { EvictParentRuntime(device, previous); } finally { oldGate.Release(); }
}
}
var rmwLock = device.GetRmwLock(parentName);
await rmwLock.WaitAsync(ct).ConfigureAwait(false);
try
@@ -1176,7 +1244,11 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
var parentRuntime = await EnsureParentRuntimeAsync(device, parentName, ct).ConfigureAwait(false);
await parentRuntime.ReadAsync(ct).ConfigureAwait(false);
var readStatus = parentRuntime.GetStatus();
if (readStatus != 0) return AbLegacyStatusMapper.MapStatus(readStatus);
if (readStatus != 0)
{
if (IsNoReply((AbStatus)readStatus)) EvictParentRuntime(device, parentName);
return (AbLegacyStatusMapper.MapStatus(readStatus), readStatus);
}
// Mask the decoded word to its native width: DecodeValue for a 16-bit Int returns a
// sign-extended int, so a word with bit 15 set decodes negative. Masking pins the
@@ -1193,9 +1265,8 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
parentRuntime.EncodeValue(AbLegacyDataType.Int, bitIndex: null, (short)updated);
await parentRuntime.WriteAsync(ct).ConfigureAwait(false);
var writeStatus = parentRuntime.GetStatus();
return writeStatus == 0
? AbLegacyStatusMapper.Good
: AbLegacyStatusMapper.MapStatus(writeStatus);
if (writeStatus != 0 && IsNoReply((AbStatus)writeStatus)) EvictParentRuntime(device, parentName);
return (writeStatus == 0 ? AbLegacyStatusMapper.Good : AbLegacyStatusMapper.MapStatus(writeStatus), writeStatus);
}
finally
{
@@ -1381,7 +1452,7 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
/// (<see cref="ConnectionThrottledException"/>). The HOT path (cached handle) never touches the gate. The
/// caller records the outcome on the device's backoff (once — recording here as well would double-count).
/// </summary>
private static async Task<IAbLegacyTagRuntime> CreateSingleFlightAsync(
private async Task<IAbLegacyTagRuntime> CreateSingleFlightAsync(
DeviceState device,
System.Collections.Concurrent.ConcurrentDictionary<string, IAbLegacyTagRuntime> table,
string key,
@@ -1395,7 +1466,7 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
{
// Re-check under the gate: a sibling caller may have connected this very key while we waited.
if (table.TryGetValue(key, out existing)) return existing;
if (device.IsThrottled(DateTime.UtcNow))
if (device.IsThrottled(Clock()))
throw new ConnectionThrottledException(
$"AbLegacy connect for '{key}' on {device.Options.HostAddress} throttled: the device is in its reconnect backoff window.");
@@ -1544,7 +1615,7 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
/// </summary>
public void Dispose()
{
var retired = RetireTables(bumpEpoch: true);
var (retired, teardownEpoch) = RetireTables(bumpEpoch: true);
// ValueTask.ConfigureAwait(false).GetAwaiter().GetResult() — drains the poll engine
// shutdown on the current thread without capturing the SynchronizationContext. The
// engine cancels every loop CTS up-front then either completes immediately (no
@@ -1557,8 +1628,12 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
JoinProbeLoopsSync(retired);
foreach (var state in retired) state.Retire();
_resolver.Clear(); // drop transient equipment-tag parses so a config change re-parses
// A newer InitializeAsync that ran while this teardown was still joining owns health now.
lock (_lifecycleLock)
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
{
if (teardownEpoch == _lifecycleEpoch)
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
}
}
/// <summary>
@@ -1620,6 +1695,11 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
public System.Collections.Concurrent.ConcurrentDictionary<string, string> IndirectLastName { get; } =
new(StringComparer.OrdinalIgnoreCase);
/// <summary>Indirect bit-write tag name to the parent word it last read-modify-wrote, so a moved index releases
/// the previous parent handle.</summary>
public System.Collections.Concurrent.ConcurrentDictionary<string, string> IndirectLastParent { get; } =
new(StringComparer.OrdinalIgnoreCase);
/// <summary>
/// Per-device connect gate (review 07 P7): every COLD runtime create + initialise on this device runs under
/// it, so connects are single-flight per device — a down device sees one attempt at a time, not one per