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

fix(galaxy,s7): R2 review minors — subscribe-once deadline, probe vs backoff, write-fault ownership, null array elements, write opt-in

Browse Source 
- Galaxy: deadline bounds the wait, not the subscribe RPC; BadTimeout for all tags and late handles unsubscribed; doc comment relocated
- S7: probe connect runs on the connect deadline so a timeout opens the backoff window
- S7: a successful write only retires a Faulted the write side owns
- S7: null array elements rejected (BadTypeMismatch); multi-PDU array writes documented non-atomic
- S7 integration: array write test gated on S7_SIM_ALLOW_WRITES=1

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-08 07:31:22 -04:00
parent ca0f1e7735
commit e77bca385a
9 changed files with 208 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs
+31 -6
View File
@@ -812,6 +812,9 @@ public sealed class GalaxyDriver
return ReadViaSubscribeOnceAsync(dialledRefs, cancellationToken);
}
/// <summary>Upper bound on one subscribe-once read; internal so tests can shrink it.</summary>
internal TimeSpan ReadViaSubscribeDeadline { get; set; } = TimeSpan.FromSeconds(30);
/// <summary>
/// Production read path. MxAccess has no one-shot Read RPC — every value comes
/// through the event stream. We synthesise a Read by:
@@ -826,9 +829,6 @@ public sealed class GalaxyDriver
/// caller's cancellation token fires, return a Bad-status snapshot in input order
/// so the caller still sees one snapshot per requested reference.
/// </summary>
/// <summary>Upper bound on one subscribe-once read; internal so tests can shrink it.</summary>
internal TimeSpan ReadViaSubscribeDeadline { get; set; } = TimeSpan.FromSeconds(30);
private async Task<IReadOnlyList<DataValueSnapshot>> ReadViaSubscribeOnceAsync(
IReadOnlyList<string> fullReferences, CancellationToken callerToken)
{
@@ -868,11 +868,36 @@ public sealed class GalaxyDriver
var bufferedIntervalMs = _options.MxAccess.PublishingIntervalMs;
IReadOnlyList<SubscribeResult> results;
// The RPC gets only the CALLER's token: the internal deadline bounds our wait, not the RPC, so a slow
// gateway can never surface an OperationCanceledException to a caller that did not cancel, and the
// handles it eventually creates are known (and unsubscribed below) rather than lost to a cancelled call.
var subscribeTask = _subscriber!.SubscribeBulkAsync(fullReferences, bufferedIntervalMs, callerToken);
try
{
results = await _subscriber!
.SubscribeBulkAsync(fullReferences, bufferedIntervalMs, cancellationToken)
.ConfigureAwait(false);
results = await subscribeTask.WaitAsync(cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException) when (!callerToken.IsCancellationRequested)
{
// Internal deadline fired while the subscribe RPC was still in flight.
pump.OnDataChange -= handler;
var subscriber = _subscriber!;
var logger = _logger;
_ = subscribeTask.ContinueWith(t =>
{
if (!t.IsCompletedSuccessfully) return Task.CompletedTask;
var late = t.Result.Where(r => r.WasSuccessful && r.ItemHandle > 0).Select(r => r.ItemHandle).ToArray();
return late.Length == 0
? Task.CompletedTask
: subscriber.UnsubscribeBulkAsync(late, CancellationToken.None).ContinueWith(
u => logger.LogWarning(u.Exception,
"GalaxyDriver.ReadViaSubscribeOnceAsync late UnsubscribeBulk failed for {Count} handle(s).", late.Length),
TaskContinuationOptions.OnlyOnFaulted);
}, TaskScheduler.Default).Unwrap();
var now = DateTime.UtcNow;
var timedOut = new DataValueSnapshot[fullReferences.Count];
for (var i = 0; i < timedOut.Length; i++)
timedOut[i] = new DataValueSnapshot(null, StatusCodeMap.BadTimeout, null, now);
return timedOut;
}
catch
{
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs
+19 -5
View File
@@ -1869,6 +1869,7 @@ public sealed class S7Driver
if (!outcome.HasObservations) return;
var worst = outcome.WorstFailure;
var priorWriteFault = _writeFault;
if (worst == DriverState.Faulted)
_writeFault = (DriverState.Faulted, outcome.WorstError!);
else if (outcome.AnySuccess)
@@ -1878,7 +1879,10 @@ public sealed class S7Driver
{
// Success-only batch: no health write, except to retire a Faulted that WE (the write
// side) were holding — a clean write is the only evidence the alert has cleared.
if (outcome.AnySuccess && _writeFault is null && Health.State == DriverState.Faulted)
// Only a Faulted the write side owns: a read-side Faulted (e.g. access denied on another DB) is the
// read pass's to clear, and Health still carries the write fault's own message iff it is ours.
if (outcome.AnySuccess && _writeFault is null && priorWriteFault is { } owned
&& Health.State == DriverState.Faulted && Health.LastError == owned.Message)
Health = new DriverHealth(DriverState.Healthy, Health.LastSuccessfulRead, null);
return;
}
@@ -2004,6 +2008,9 @@ public sealed class S7Driver
var elements = new List<object?>(count);
foreach (var e in items) elements.Add(e);
// Convert.ToXxx(null) is 0 — a null element must not silently write zero to the PLC.
if (elements.Contains(null))
throw new S7WriteValueException($"S7 array tag '{tag.Name}': an element is null.");
if (elements.Count != count)
throw new S7WriteValueException(
$"S7 array tag '{tag.Name}' has {count} elements but the write supplied {elements.Count}.");
@@ -2361,10 +2368,10 @@ public sealed class S7Driver
// Probe via S7.Net's low-cost GetCpuStatus — returns the CPU state (Run/Stop)
// and is intentionally light on the comms mailbox. Single-word Plc.ReadAsync
// would also work but GetCpuStatus doubles as a "PLC actually up" check.
using var probeCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
probeCts.CancelAfter(_options.Probe.Timeout);
using var gateCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
gateCts.CancelAfter(_options.Probe.Timeout);
await _gate.WaitAsync(probeCts.Token).ConfigureAwait(false);
await _gate.WaitAsync(gateCts.Token).ConfigureAwait(false);
try
{
try
@@ -2372,7 +2379,14 @@ public sealed class S7Driver
// EnsureConnectedAsync doubles as the reconnect backstop: if a read/write
// marked the handle dead (or the socket dropped between ticks), the probe
// reopens it here so recovery doesn't wait for the next data call.
var plc = await EnsureConnectedAsync(probeCts.Token).ConfigureAwait(false);
// It runs on the LOOP token, not the (shorter) probe deadline: the connect is bounded by
// _options.Timeout inside, and a timeout there must count as a failure so STAB-21's
// backoff window opens — cutting it with the probe deadline looked like a caller cancel,
// recorded nothing, and let a black-holed host cost a connect attempt every probe tick.
var plc = await EnsureConnectedAsync(ct).ConfigureAwait(false);
// The probe deadline bounds only the CPU-status round trip.
using var probeCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
probeCts.CancelAfter(_options.Probe.Timeout);
await plc.ReadStatusAsync(probeCts.Token).ConfigureAwait(false);
success = true;
}