diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs
index e1924364..60e989ce 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs
@@ -218,8 +218,18 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
///
public static readonly TimeSpan DefaultInitializeTimeout = TimeSpan.FromSeconds(30);
+ ///
+ /// Pessimistic bound on one write round-trip, on top of the 5 s token-based write cap. A driver whose
+ /// ignores its token would otherwise hold its slot AND its tag forever,
+ /// piling every later write to that tag up in the queue. 2× the write cap.
+ ///
+ public static readonly TimeSpan DefaultWriteTimeout = TimeSpan.FromSeconds(10);
+
/// Self-sent by a write task once it has replied, releasing its slot and its tag.
- private sealed record WriteCompleted(string TagId);
+ /// The tag the write held.
+ /// The actor incarnation that launched it — a late completion from a previous
+ /// incarnation (after a supervision restart) must not free the new incarnation's slot or tag.
+ private sealed record WriteCompleted(string TagId, int Incarnation);
private readonly IDriver _driver;
@@ -266,6 +276,13 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// See .
private readonly TimeSpan _initializeTimeout;
+ /// See .
+ private readonly TimeSpan _writeTimeout;
+
+ /// This actor object's identity for its own off-mailbox completions (drawn from the process-wide
+ /// generation source, so it is unique across supervision restarts of the same child).
+ private readonly int _incarnation;
+
///
/// Set by so the it triggers skips the dispose. A
/// supervision restart recreates the actor object around the SAME (the Props closure
@@ -407,6 +424,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// BadTimeout; defaults to .
/// Optional pessimistic bound on one initialise attempt; defaults to
/// .
+ /// Optional pessimistic bound on one write round-trip; defaults to
+ /// .
/// A instance configured to create the wrapped .
public static Props Props(
IDriver driver,
@@ -420,7 +439,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
TimeSpan? shutdownTimeout = null,
int? maxWritesInFlight = null,
TimeSpan? queuedWriteBudget = null,
- TimeSpan? initializeTimeout = null) =>
+ TimeSpan? initializeTimeout = null,
+ TimeSpan? writeTimeout = null) =>
Akka.Actor.Props.Create(() => new DriverInstanceActor(
driver,
reconnectInterval ?? DefaultReconnectInterval,
@@ -433,7 +453,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
shutdownTimeout,
maxWritesInFlight,
queuedWriteBudget,
- initializeTimeout));
+ initializeTimeout,
+ writeTimeout));
///
/// Returns true when the driver should boot in DEV-STUB mode based on host platform and
@@ -478,6 +499,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
/// when null or non-positive.
/// Pessimistic bound on one initialise attempt; defaults to
/// when null or non-positive.
+ /// Pessimistic bound on one write round-trip; defaults to
+ /// when null or non-positive.
public DriverInstanceActor(
IDriver driver,
TimeSpan reconnectInterval,
@@ -490,8 +513,11 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
TimeSpan? shutdownTimeout = null,
int? maxWritesInFlight = null,
TimeSpan? queuedWriteBudget = null,
- TimeSpan? initializeTimeout = null)
+ TimeSpan? initializeTimeout = null,
+ TimeSpan? writeTimeout = null)
{
+ _writeTimeout = writeTimeout is { } wt && wt > TimeSpan.Zero ? wt : DefaultWriteTimeout;
+ _incarnation = Interlocked.Increment(ref s_initGenerationSource);
_maxWritesInFlight = maxWritesInFlight is > 0 ? maxWritesInFlight.Value : DefaultMaxWritesInFlight;
_queuedWriteBudget = queuedWriteBudget is { } qb && qb > TimeSpan.Zero ? qb : DefaultQueuedWriteBudget;
_initializeTimeout = initializeTimeout is { } it && it > TimeSpan.Zero ? it : DefaultInitializeTimeout;
@@ -1097,19 +1123,22 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
var self = Self;
var hostKey = ResolveHost(msg.TagId);
var invoker = _invoker;
- _ = RunWriteAsync(writable, invoker, hostKey, msg, replyTo, self);
+ _ = RunWriteAsync(writable, invoker, hostKey, msg, replyTo, self, _writeTimeout, _incarnation, _log);
}
private static async Task RunWriteAsync(
IWritable writable, IDriverCapabilityInvoker invoker, string hostKey,
- WriteAttribute msg, IActorRef replyTo, IActorRef self)
+ WriteAttribute msg, IActorRef replyTo, IActorRef self, TimeSpan writeTimeout, int incarnation,
+ ILoggingAdapter log)
{
// Off the actor context from here on: the first await yields, and nothing below reads actor state.
await Task.Yield();
var request = new[] { new WriteRequest(msg.TagId, msg.Value) };
// Bound the write so a hung backend can't hold a slot forever — retry stays
- // off by default, but a stalled call still needs an answer.
- using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+ // off by default, but a stalled call still needs an answer. Disposed below, except when the write is
+ // abandoned (the orphan may still be holding the token).
+ var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+ var abandoned = false;
try
{
// Route through the resilience invoker (Write pipeline: timeout, per-host breaker). Writes
@@ -1119,11 +1148,31 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
// therefore authoritative regardless of any configured Write retry. Per-tag opt-in to retries via
// WriteIdempotentAttribute is the unshipped forward path (see WriteIdempotentAttribute). The outer
// cts is retained as a hard backstop above the (typically tighter) tier timeout.
- var results = await invoker.ExecuteWriteAsync(
+ // PESSIMISTIC bound on top of the token-based cap: a driver that ignores its token would otherwise
+ // hold the slot and the tag forever, piling every later write to that tag up in the queue.
+ var attempt = invoker.ExecuteWriteAsync(
hostKey,
isIdempotent: false,
async ct => await writable.WriteAsync(request, ct),
- cts.Token);
+ cts.Token).AsTask();
+ IReadOnlyList results;
+ try
+ {
+ results = await attempt.WaitAsync(writeTimeout);
+ }
+ catch (TimeoutException)
+ {
+ abandoned = true;
+ log.Warning(
+ "DriverInstance write of {Tag} did not complete within {Bound} (the driver ignored its cancellation " +
+ "token); abandoned — its slot and tag are released, its eventual outcome is discarded",
+ msg.TagId, writeTimeout);
+ try { cts.Cancel(); } catch (ObjectDisposedException) { }
+ _ = attempt.ContinueWith(t => _ = t.Exception, TaskContinuationOptions.OnlyOnFaulted);
+ replyTo.Tell(new WriteAttributeResult(false,
+ $"StatusCode=0x800A0000 (BadTimeout: write did not complete within {writeTimeout.TotalSeconds:F0}s)"));
+ return;
+ }
if (results is { Count: 1 } && IsGoodStatus(results[0].StatusCode))
{
replyTo.Tell(new WriteAttributeResult(true, null));
@@ -1142,14 +1191,18 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
}
finally
{
- self.Tell(new WriteCompleted(msg.TagId));
+ if (!abandoned) cts.Dispose();
+ self.Tell(new WriteCompleted(msg.TagId, incarnation));
}
}
/// Frees the write's slot and tag and drains the queue: the oldest entries whose tag is free are
- /// launched (in arrival order) while slots remain; entries older than the budget are failed instead.
+ /// launched (in arrival order) while slots remain; entries older than the budget are failed instead. A
+ /// completion stamped with another incarnation (a write launched before a supervision restart) is ignored —
+ /// it never held a slot or tag of THIS incarnation.
private void OnWriteCompleted(WriteCompleted msg)
{
+ if (msg.Incarnation != _incarnation) return;
if (_writesInFlight > 0) _writesInFlight--;
_tagsInFlight.Remove(msg.TagId);
DrainWriteQueue();
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs
index fc2f041f..f6f99002 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorConcurrentWriteTests.cs
@@ -23,12 +23,12 @@ public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBa
private static readonly TimeSpan WriteLatency = TimeSpan.FromMilliseconds(1200);
private (IActorRef Actor, Akka.TestKit.TestProbe Parent) Connect(
- IDriver driver, int? maxWritesInFlight = null, TimeSpan? queuedWriteBudget = null)
+ IDriver driver, int? maxWritesInFlight = null, TimeSpan? queuedWriteBudget = null, TimeSpan? writeTimeout = null)
{
var parent = CreateTestProbe();
var actor = parent.ChildActorOf(DriverInstanceActor.Props(
driver, commsLossTimeout: TimeSpan.Zero, maxWritesInFlight: maxWritesInFlight,
- queuedWriteBudget: queuedWriteBudget));
+ queuedWriteBudget: queuedWriteBudget, writeTimeout: writeTimeout));
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
parent.ExpectMsg(m => m.Connected, PresenceBudget);
return (actor, parent);
@@ -163,6 +163,38 @@ public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBa
driver.EnteredRefs.ShouldBe(new[] { "tag-0" }, "the stale write must never reach the device");
}
+ ///
+ /// Fix round 2: a token-ignoring hung write held its tag forever, so every later write to that tag piled up
+ /// in the queue. The write is bounded pessimistically; on overrun it answers BadTimeout and releases its
+ /// slot and tag, and a later write to the same tag reaches the device.
+ ///
+ [Fact]
+ public async Task A_hung_token_ignoring_write_is_abandoned_and_releases_its_tag_for_a_later_write()
+ {
+ var driver = new SlowWritableDriver { HoldWrites = true, IgnoreCancellation = true };
+ var (actor, _) = Connect(driver, maxWritesInFlight: 1, writeTimeout: TimeSpan.FromMilliseconds(300));
+
+ var hung = actor.Ask(
+ new DriverInstanceActor.WriteAttribute("tag-a", 1), PresenceBudget);
+ AwaitCondition(() => driver.WritesEntered == 1, PresenceBudget);
+
+ var hungResult = await hung;
+ hungResult.Success.ShouldBeFalse();
+ hungResult.Reason!.ShouldContain("800A0000", customMessage: "BadTimeout");
+ driver.MaxConcurrentWrites.ShouldBe(1);
+
+ // The tag (and the single slot) are free again: a later tag-a write reaches the device even though the
+ // abandoned one is still inside it.
+ var later = actor.Ask(
+ new DriverInstanceActor.WriteAttribute("tag-a", 2), PresenceBudget);
+ AwaitCondition(() => driver.WritesEntered == 2, PresenceBudget);
+ driver.EnteredRefs.ShouldBe(new[] { "tag-a", "tag-a" });
+ driver.MaxConcurrentWrites.ShouldBe(2, "positive control: the abandoned write really is still wedged inside the driver");
+
+ driver.ReleaseWrites();
+ (await later).Success.ShouldBeTrue();
+ }
+
[Fact]
public async Task Queued_writes_fail_fast_when_the_driver_leaves_Connected()
{
@@ -201,6 +233,8 @@ public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBa
public TimeSpan Latency { get; init; }
public bool HoldWrites { get; init; }
+ /// When true a held write ignores its cancellation token — a wedged driver.
+ public bool IgnoreCancellation { get; init; }
public IReadOnlyList Writes => _writes.ToArray();
public IReadOnlyList EnteredRefs => _enteredRefs.ToArray();
@@ -227,7 +261,8 @@ public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBa
Interlocked.CompareExchange(ref _maxInside, inside, seen);
try
{
- if (HoldWrites) await _release.Task.WaitAsync(cancellationToken);
+ if (HoldWrites && IgnoreCancellation) await _release.Task;
+ else if (HoldWrites) await _release.Task.WaitAsync(cancellationToken);
else if (Latency > TimeSpan.Zero) await Task.Delay(Latency, cancellationToken);
foreach (var w in writes) _writes.Enqueue(w);
return writes.Select(_ => new WriteResult(OpcUaStatus.Good)).ToList();