fix(focas): PMC coalescing treats any rc refusal as per-tag fallback, skips transient busy memo, one connect per batch (#266)

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-08 17:03:25 -04:00
parent 487d394cde
commit c86204b7d6
4 changed files with 99 additions and 21 deletions
@@ -634,6 +634,20 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
{ {
if (items.Count < 2) continue; if (items.Count < 2) continue;
var plans = FocasPmcCoalescer.Plan(items, _options.PmcCoalesce.MaxSpanBytes, _options.PmcCoalesce.MaxGapBytes); var plans = FocasPmcCoalescer.Plan(items, _options.PmcCoalesce.MaxSpanBytes, _options.PmcCoalesce.MaxGapBytes);
if (!plans.Any(p => p.Members.Count >= 2 && !device.CoalesceRejected.ContainsKey(p.Key))) continue;
// One connect attempt per device per batch: if it fails, every PMC tag of the device fails with
// that status here instead of paying a connect attempt per plan and again per tag.
IFocasClient client;
try { client = await EnsureConnectedAsync(device, ct).ConfigureAwait(false); }
catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; }
catch (Exception)
{
FailMembers(items, FocasStatusMapper.BadCommunicationError, references, results, now, ref served,
$"FOCAS connect failed reading PMC tags of {device.Options.HostAddress}");
continue;
}
foreach (var plan in plans) foreach (var plan in plans)
{ {
if (plan.Members.Count < 2) continue; if (plan.Members.Count < 2) continue;
@@ -642,7 +656,6 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
var members = plan.Members; var members = plan.Members;
try try
{ {
var client = await EnsureConnectedAsync(device, ct).ConfigureAwait(false);
var (slots, status) = await client var (slots, status) = await client
.ReadPmcSlotsAsync(plan.Letter, plan.SlotType, plan.Start, plan.End, ct) .ReadPmcSlotsAsync(plan.Letter, plan.SlotType, plan.Start, plan.End, ct)
.ConfigureAwait(false); .ConfigureAwait(false);
@@ -659,44 +672,41 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
} }
Volatile.Write(ref _health, new DriverHealth(DriverState.Healthy, now, null)); Volatile.Write(ref _health, new DriverHealth(DriverState.Healthy, now, null));
} }
else if (status == FocasStatusMapper.BadCommunicationError)
{
FailMembers(plan, status, references, results, now, ref served);
}
else else
{ {
// The CNC refused the merged range. Don't guess which member is at fault: remember the // The CNC REFUSED the merged range (any return code, mapped or not). Don't guess which
// range and let the per-tag loop give each member its own answer. // member is at fault: let the per-tag loop give each its own answer. Remember the range
device.CoalesceRejected[plan.Key] = 0; // so later polls skip the doomed read — except for a transient refusal (EW_BUSY maps to
// BadDeviceFailure), which says nothing about the range.
if (status != FocasStatusMapper.BadDeviceFailure)
device.CoalesceRejected[plan.Key] = 0;
_logger.LogDebug( _logger.LogDebug(
"FOCAS PMC range {Key} on {Host} rejected (0x{Status:X8}) — reading its {Count} tags individually", "FOCAS PMC range {Key} on {Host} rejected (0x{Status:X8}) — reading its {Count} tags individually",
plan.Key, device.Options.HostAddress, status, members.Count); plan.Key, device.Options.HostAddress, status, members.Count);
} }
} }
catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; } catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; }
catch (OperationCanceledException)
{
FailMembers(plan, FocasStatusMapper.BadCommunicationError, references, results, now, ref served);
}
catch (Exception) catch (Exception)
{ {
FailMembers(plan, FocasStatusMapper.BadCommunicationError, references, results, now, ref served); // Transport failure / timeout: the link is suspect, so fail the members without per-tag retries.
FailMembers(members, FocasStatusMapper.BadCommunicationError, references, results, now, ref served,
$"FOCAS transport failure reading PMC range {plan.Key}");
} }
} }
} }
return served; return served;
void FailMembers(FocasPmcRangePlan plan, uint status, IReadOnlyList<string> refs, void FailMembers(IEnumerable<FocasPmcReadItem> failed, uint status, IReadOnlyList<string> refs,
DataValueSnapshot[] res, DateTime stamp, ref bool[]? servedRef) DataValueSnapshot[] res, DateTime stamp, ref bool[]? servedRef, string reason)
{ {
foreach (var m in plan.Members) foreach (var m in failed)
{ {
res[m.Index] = new DataValueSnapshot(null, status, null, stamp); res[m.Index] = new DataValueSnapshot(null, status, null, stamp);
(servedRef ??= new bool[refs.Count])[m.Index] = true; (servedRef ??= new bool[refs.Count])[m.Index] = true;
} }
Volatile.Write(ref _health, new DriverHealth(DriverState.Degraded, Volatile.Write(ref _health, new DriverHealth(DriverState.Degraded,
Volatile.Read(ref _health).LastSuccessfulRead, Volatile.Read(ref _health).LastSuccessfulRead,
$"FOCAS status 0x{status:X8} reading PMC range {plan.Key}")); $"{reason} (0x{status:X8})"));
} }
} }
@@ -84,7 +84,10 @@ public interface IFocasClient : IDisposable
/// <param name="start">First byte address.</param> /// <param name="start">First byte address.</param>
/// <param name="end">Last byte address, inclusive; a whole number of slots from <paramref name="start"/>.</param> /// <param name="end">Last byte address, inclusive; a whole number of slots from <paramref name="start"/>.</param>
/// <param name="cancellationToken">The cancellation token.</param> /// <param name="cancellationToken">The cancellation token.</param>
/// <returns>The raw slot values (null when not Good) and the mapped OPC UA status.</returns> /// <returns>
/// The raw slot values (null when not Good) and the mapped OPC UA status. A returned non-Good status is
/// always a CNC return-code refusal; a transport failure or timeout throws instead.
/// </returns>
Task<(IReadOnlyList<long>? slots, uint status)> ReadPmcSlotsAsync( Task<(IReadOnlyList<long>? slots, uint status)> ReadPmcSlotsAsync(
string letter, FocasDataType slotType, int start, int end, CancellationToken cancellationToken); string letter, FocasDataType slotType, int start, int end, CancellationToken cancellationToken);
@@ -71,7 +71,10 @@ public sealed class WireFocasClient : IFocasClient
public async Task<(IReadOnlyList<long>? slots, uint status)> ReadPmcSlotsAsync( public async Task<(IReadOnlyList<long>? slots, uint status)> ReadPmcSlotsAsync(
string letter, FocasDataType slotType, int start, int end, CancellationToken cancellationToken) string letter, FocasDataType slotType, int start, int end, CancellationToken cancellationToken)
{ {
if (!_wire.IsConnected) return (null, FocasStatusMapper.BadCommunicationError); // Contract: a RETURNED status is always a CNC return code (a refusal); a transport failure THROWS.
// The coalescing caller needs that distinction — it falls back per tag on a refusal but must not
// re-hit a dead link (#266).
RequireConnected();
cancellationToken.ThrowIfCancellationRequested(); cancellationToken.ThrowIfCancellationRequested();
var area = FocasPmcAreaLookup.FromLetter(letter); var area = FocasPmcAreaLookup.FromLetter(letter);
@@ -85,9 +88,9 @@ public sealed class WireFocasClient : IFocasClient
return (null, FocasStatusMapper.MapFocasReturn(result.Rc)); return (null, FocasStatusMapper.MapFocasReturn(result.Rc));
return (result.Value.Values, FocasStatusMapper.Good); return (result.Value.Values, FocasStatusMapper.Good);
} }
catch (FocasWireException ex) catch (FocasWireException ex) when (ex.Rc is short rc)
{ {
return (null, ex.Rc is short rc ? FocasStatusMapper.MapFocasReturn(rc) : FocasStatusMapper.BadCommunicationError); return (null, FocasStatusMapper.MapFocasReturn(rc));
} }
} }
@@ -132,6 +132,68 @@ public sealed class FocasPmcCoalescingDriverTests
await drv.ShutdownAsync(CancellationToken.None); await drv.ShutdownAsync(CancellationToken.None);
} }
[Fact]
public async Task A_refusal_with_an_unmapped_rc_that_maps_to_BadCommunicationError_still_falls_back_per_tag()
{
var (drv, c) = await StartAsync(null,
Tag("A", "R100", FocasDataType.Byte), Tag("B", "R101", FocasDataType.Byte));
c.PmcRangeRejectors.Add((_, _, _) => true);
c.PmcRangeRejectStatus = FocasStatusMapper.BadCommunicationError; // what an unlisted rc / EW_PARITY / EW_SOCKET maps to
c.Values["R100"] = (sbyte)1; c.Values["R101"] = (sbyte)2;
var r = await drv.ReadAsync(["A", "B"], CancellationToken.None);
r.Select(s => s.Value).ShouldBe([(object)(sbyte)1, (sbyte)2]);
r.ShouldAllBe(s => s.StatusCode == FocasStatusMapper.Good);
await drv.ReadAsync(["A", "B"], CancellationToken.None);
c.PmcRangeReads.Count.ShouldBe(1, "a deterministic rc refusal is memoised");
await drv.ShutdownAsync(CancellationToken.None);
}
[Fact]
public async Task A_transient_busy_refusal_is_not_memoised()
{
var (drv, c) = await StartAsync(null,
Tag("A", "R100", FocasDataType.Byte), Tag("B", "R101", FocasDataType.Byte));
var busy = true;
c.PmcRangeRejectors.Add((_, _, _) => busy);
c.PmcRangeRejectStatus = FocasStatusMapper.BadDeviceFailure; // EW_BUSY
c.Values["R100"] = (sbyte)1; c.Values["R101"] = (sbyte)2;
(await drv.ReadAsync(["A", "B"], CancellationToken.None)).ShouldAllBe(s => s.StatusCode == FocasStatusMapper.Good);
busy = false;
c.PmcMemory[("R", 100)] = 5; c.PmcMemory[("R", 101)] = 6;
var r = await drv.ReadAsync(["A", "B"], CancellationToken.None);
c.PmcRangeReads.Count.ShouldBe(2, "the range is retried once the CNC is no longer busy");
r.Select(s => s.Value).ShouldBe([(object)(sbyte)5, (sbyte)6]);
await drv.ShutdownAsync(CancellationToken.None);
}
[Fact]
public async Task A_down_device_costs_one_connect_attempt_per_batch_not_one_per_plan_and_tag()
{
var client = new FakeFocasClient { ThrowOnConnect = true };
var drv = new FocasDriver(new FocasDriverOptions
{
Devices = [new FocasDeviceOptions(Host)],
Probe = new FocasProbeOptions { Enabled = false },
RawTags = FocasRawTags.Entries(new[]
{
Tag("A", "R100", FocasDataType.Byte), Tag("B", "R101", FocasDataType.Byte),
Tag("C", "R900", FocasDataType.Byte), Tag("D", "R2000", FocasDataType.Byte),
}),
}, "drv-1", new FakeFocasClientFactory { Customise = () => client });
await drv.InitializeAsync("{}", CancellationToken.None);
var before = client.ConnectCount;
var r = await drv.ReadAsync(["A", "B", "C", "D"], CancellationToken.None);
r.ShouldAllBe(s => s.StatusCode == FocasStatusMapper.BadCommunicationError);
(client.ConnectCount - before).ShouldBe(1);
await drv.ShutdownAsync(CancellationToken.None);
}
[Fact] [Fact]
public async Task A_transport_failure_fails_every_member_without_per_tag_retries() public async Task A_transport_failure_fails_every_member_without_per_tag_retries()
{ {