Files
lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/ActorNodeWriteGateway.cs
T
Joseph Doherty 97dc715dac
v2-ci / unit-tests (push) Has been cancelled
fix(runtime): derive every write bound and Ask from one write timeout (lmxopcua#777)
Three write bounds were configured independently: the inbound-write
gateway's 10 s Ask, DriverHostActor's fixed 8 s Ask of the child, and the
child's own write timeout (5 s + up to 2 s grace) with a fixed 8 s queue
budget. An outer Ask that fires first reports the write failed and the node
reverts while the child is still driving the device, so any write timeout
above ~6 s (or a write dequeued late in its 8 s budget) let the device take
a value the client was told failed.

DriverWriteBounds (Runtime) derives them all from DriverWrites:Timeout:
  token cancel   = Timeout
  abandon        = Timeout + min(Timeout/2, 2 s)            (RunBound)
  queue budget   = RunBound
  child replies  <= queue budget + RunBound                  (ReplyBound)
  host Ask       = ReplyBound + 1 s
  gateway Ask    = host Ask + 1 s
Default Timeout is 10 s (the Tier C Write timeout, so no tier default is
capped below its own policy); host Ask 25 s, gateway Ask 26 s. Tier A/B
writes are still cancelled first by their 2 s / 4 s tier timeouts.

- DriverHostActor takes writeBounds + writeClock, spawns children with them,
  and runs its Ask deadline as a token on the write clock (Infinite Akka
  timeout), so host and child bounds share one clock.
- WriteAttribute gains IssuedAt (write-clock timestamp, stamped by the
  host): the queue budget counts mailbox time too, and a write past it is
  failed BadTimeout and never started - including on arrival. Stale queued
  writes are expired across the whole queue, not only up to a free slot.
- ActorNodeWriteGateway's default Ask = Default.GatewayAsk; the hosted
  service passes the configured one.
- Startup validation (Host DriverWriteBoundsBinding, same warn-and-clamp
  style as ScriptingOptionsBinding): unparsable / non-positive -> default,
  above 5 min -> clamped, and each tier whose default Write timeout exceeds
  the write timeout is named in a Warning (its policy would never take
  effect). The derived bounds are logged at Information. Per-instance
  ResilienceConfig Write overrides arrive with a deployment, so
  DriverCapabilityInvokerFactory warns at spawn (writeTimeoutCeiling).
- appsettings.json documents DriverWrites:Timeout.

Tests (manual clock):
- DriverHostActorWriteAskTests: a held write under a 9 s timeout survives
  an 8.5 s step and succeeds; a token-ignoring write is answered by the
  child's BadTimeout at 9 s + grace, not by the Ask; the Infinite-timeout
  Ask ends exactly at its clock deadline.
- DriverInstanceActorConcurrentWriteTests: a write issued longer ago than
  the queue budget is failed and never reaches the driver; positive control
  at exactly the budget is written.
- DriverWriteBoundsTests: every outer bound covers the inner ones for 0.5 s
  to 300 s timeouts.
- DriverWriteBoundsBindingTests: default fits every tier; 3 s timeout is
  caught at startup with Tier B and C named (not A); bad values fall back
  with a warning; over-max clamps.
- DriverCapabilityInvokerFactoryTests: spawn warns when a Write override
  exceeds the ceiling; no warning within it or without a ceiling.
Mutation checks: a fixed 8 s host Ask on the same clock fails both 9 s host
tests; stamping arrival time instead of IssuedAt fails the issued-at test;
dropping the tier check / factory check fails their tests.

Runtime.Tests 961 passed / 3 skipped at AKKA_TEST_TIMEFACTOR 3 and 1;
Host.Tests 232/232 at 3 and 1; Core.Tests 361 passed / 1 skipped.
Solution Release build: 0 warnings.

ContinuousHistorizationRecorderTypedValueTests (historian) deliberately
left for the historian lane.

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
2026-11-09 15:46:30 -05:00

81 lines
4.1 KiB
C#

using Akka.Actor;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
/// <summary>
/// Akka adapter for the Commons <see cref="IOpcUaNodeWriteGateway"/>: routes an inbound OPC UA
/// operator write to the local <see cref="DriverHostActor"/> by Asking it a
/// <see cref="DriverHostActor.RouteNodeWrite"/> and translating the reply
/// <see cref="DriverHostActor.NodeWriteResult"/> into a <see cref="NodeWriteOutcome"/>.
///
/// <para>
/// The node manager calls <see cref="WriteAsync"/> fire-and-forget from its OnWriteValue handler,
/// which runs under the node-manager Lock, so the method does no blocking work before its first
/// await — resolving the actor and building the Ask returns a Task promptly. The
/// <see cref="DriverHostActor"/> reference is resolved <em>lazily per write</em> via
/// <c>resolveDriverHost</c>: the host wires this gateway during StartAsync, before the Akka
/// <see cref="DriverHostActor"/> registers, so a one-shot resolve at construction would always miss
/// and leave every write unavailable. By write time (long after startup) the registry has it.
/// </para>
/// </summary>
public sealed class ActorNodeWriteGateway : IOpcUaNodeWriteGateway
{
private readonly Func<IActorRef?> _resolveDriverHost;
private readonly TimeSpan _askTimeout;
private readonly ILogger _logger;
/// <summary>Creates the gateway.</summary>
/// <param name="resolveDriverHost">Lazy per-write resolver for the local <see cref="DriverHostActor"/>;
/// returns null until the actor has registered (StartAsync ordering — the actor registers AFTER the host
/// wires this gateway).</param>
/// <param name="logger">Logger for dropped/rejected/timed-out writes.</param>
/// <param name="askTimeout">Ask timeout; defaults to <see cref="DriverWriteBounds.GatewayAsk"/> of
/// <see cref="DriverWriteBounds.Default"/>. Production passes the GatewayAsk of the node's configured write
/// bounds, so this Ask always outlasts the host's (lmxopcua#777) — an outer Ask that fires first reports the
/// write failed, and the node reverts, while the device may still take the value.</param>
public ActorNodeWriteGateway(Func<IActorRef?> resolveDriverHost, ILogger logger, TimeSpan? askTimeout = null)
{
_resolveDriverHost = resolveDriverHost ?? throw new ArgumentNullException(nameof(resolveDriverHost));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_askTimeout = askTimeout ?? DriverWriteBounds.Default.GatewayAsk;
}
/// <inheritdoc />
public async Task<NodeWriteOutcome> WriteAsync(string nodeId, object? value, AddressSpaceRealm realm, CancellationToken ct)
{
var driverHost = _resolveDriverHost();
if (driverHost is null)
{
_logger.LogWarning("Inbound write to {NodeId} dropped: no DriverHostActor registered", nodeId);
return new NodeWriteOutcome(false, "writes unavailable");
}
try
{
var result = await driverHost.Ask<DriverHostActor.NodeWriteResult>(
new DriverHostActor.RouteNodeWrite(nodeId, value, realm), _askTimeout, ct).ConfigureAwait(false);
if (!result.Success)
_logger.LogWarning("Operator write to {NodeId} rejected: {Reason}", nodeId, result.Reason);
return new NodeWriteOutcome(result.Success, result.Reason);
}
catch (OperationCanceledException ex)
{
_logger.LogWarning(ex, "Operator write to {NodeId} cancelled", nodeId);
return new NodeWriteOutcome(false, "write cancelled");
}
catch (AskTimeoutException ex)
{
_logger.LogWarning(ex, "Operator write to {NodeId} timed out", nodeId);
return new NodeWriteOutcome(false, "write timeout");
}
catch (Exception ex)
{
_logger.LogError(ex, "Operator write to {NodeId} failed unexpectedly", nodeId);
return new NodeWriteOutcome(false, "write error");
}
}
}