diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs
index e66d7a8d..ecd77f80 100644
--- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs
+++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs
@@ -704,10 +704,24 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
{
ok = await _writer.WriteLiveValuesAsync(g.Key, g.Values, ct).ConfigureAwait(false);
}
+ catch (OperationCanceledException) when (ct.IsCancellationRequested)
+ {
+ // Shutdown cut this write short. That is a property of the PASS, not of the tag:
+ // swallowed as `ok = false` it becomes a Gitea #538 consecutive failure, and when a
+ // sibling group already wrote this pass the fold books it — so a routine node stop
+ // can tip a perfectly HEALTHY tag over its attempt budget and discard its queued
+ // samples on the way out. Rethrow so it propagates through Task.WhenAll to the
+ // method's outer cancellation catch, which abandons the pass with no ack and no
+ // counters touched; the entries stay durable and drain next startup.
+ throw;
+ }
catch
{
// The writer is non-throwing by contract; guard defensively so one tag's fault
// cannot fail the whole drain task (and, now, cannot fault the WhenAll either).
+ // A cancellation-SHAPED exception raised while our token is still live (a gateway's
+ // own internal deadline, say) lands here deliberately: it is an ordinary write
+ // failure, and exempting it would create a tag #538 can never quarantine.
ok = false;
}
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs
index b4f44cca..615e4010 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs
@@ -524,6 +524,145 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
}, TimeSpan.FromSeconds(10));
}
+ // ------- Shutdown cancellation is a pass ABORT, not a per-tag failure (guards #538) -------
+
+ ///
+ /// A write that is cut short by the drain token's cancellation (node shutdown) must abort the
+ /// whole pass, NOT be booked against that tag as a Gitea #538 consecutive failure.
+ ///
+ ///
+ /// The per-tag lambda wrapped the writer call in a bare catch, so a shutdown
+ /// was converted to "this tag failed to write". When a
+ /// sibling tag had already written in the same pass, the #538 quarantine counted it — so a
+ /// routine shutdown could push a perfectly HEALTHY tag over its attempt budget and DISCARD its
+ /// queued samples on the way out. The outer
+ /// catch (OperationCanceledException) when (ct.IsCancellationRequested) already treats a
+ /// cancelled pass as a non-fault, but the inner catch ate the exception before it could be reached.
+ ///
+ /// The load-bearing assertion is that BOTH entries are still queued afterwards: the pass
+ /// neither acked (the healthy tag's entry survives) nor abandoned (the cancelled tag's entry
+ /// survives). Against the unfixed recorder the failing tag goes one over budget on the
+ /// cancelled pass and the whole peeked prefix is truncated away, so the outbox reads 0.
+ ///
+ ///
+ /// The budget is deliberately the tightest there is (maxWriteAttempts: 1) with exactly
+ /// one genuine failure banked first, because a cancelled pass can only ever be observed to
+ /// abandon when it is the attempt that tips the tag OVER its budget — the counter is clamped
+ /// to at least 1, so a first-ever cancelled pass is retained either way and would prove nothing.
+ ///
+ ///
+ [Fact]
+ public async Task A_shutdown_cancelled_write_aborts_the_pass_without_counting_a_tag_failure()
+ {
+ var mux = CreateTestProbe();
+ // "Bad.Tag" fails once for real, then BLOCKS — so the shutdown lands mid-write on the pass that
+ // would otherwise tip it over the budget.
+ var writer = new ShutdownBlockingValueWriter("Bad.Tag", failCalls: 1);
+ var outbox = new InMemoryOutbox();
+
+ IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props(
+ mux.Ref, writer, outbox, historizedRefs: new[] { "Good.Tag", "Bad.Tag" },
+ minBackoff: TimeSpan.FromMilliseconds(20),
+ maxBackoff: TimeSpan.FromMilliseconds(50),
+ drainInterval: TimeSpan.FromMilliseconds(20),
+ maxWriteAttempts: 1,
+ // Sequential, so the batch's FIRST group (the healthy tag) has demonstrably written before
+ // the second one blocks — "a sibling succeeded this pass" is the #538 discriminator, and
+ // this test is only meaningful when it is true.
+ drainConcurrency: 1));
+
+ rec.Tell(new VirtualTagActor.DependencyValueChanged("Good.Tag", 1.0, DateTime.UtcNow));
+ rec.Tell(new VirtualTagActor.DependencyValueChanged("Bad.Tag", 2.0, DateTime.UtcNow));
+
+ await writer.Entered.WaitAsync(TimeSpan.FromSeconds(10));
+
+ // Preconditions, so a green result can never be vacuous: both entries are still queued (they were
+ // peeked as ONE batch — a lone healthy entry would have been acked and gone), and the healthy tag
+ // has written on this very pass.
+ Assert.Equal(2, await outbox.CountAsync(default));
+ Assert.Equal(2, writer.CallCountFor("Good.Tag"));
+
+ Sys.Stop(rec); // PostStop cancels the drain token underneath the in-flight write
+
+ await writer.Cancelled.WaitAsync(TimeSpan.FromSeconds(10));
+
+ // Give the (unfixed) fold every chance to book the failure and truncate before we assert it did not.
+ await Task.Delay(TimeSpan.FromMilliseconds(500));
+ Assert.Equal(2, await outbox.CountAsync(default));
+ }
+
+ ///
+ /// Scope pin: a NON-cancellation throw from one tag, while a sibling wrote, still counts toward
+ /// the Gitea #538 budget and still abandons.
+ ///
+ ///
+ /// The fix narrows the per-tag catch by one exception shape only. A writer that faults for any
+ /// other reason is exactly the poison tag #538 exists for, and must keep pinning nothing.
+ ///
+ [Fact]
+ public async Task A_non_cancellation_throw_still_counts_toward_abandonment()
+ {
+ var mux = CreateTestProbe();
+ var writer = new FakeValueWriter { Succeed = true };
+ writer.SetThrowing("Bad.Tag", static _ => new InvalidOperationException("gateway write faulted"));
+ var outbox = new InMemoryOutbox();
+
+ IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props(
+ mux.Ref, writer, outbox, historizedRefs: new[] { "Good.Tag", "Bad.Tag" },
+ minBackoff: TimeSpan.FromMilliseconds(20),
+ maxBackoff: TimeSpan.FromMilliseconds(50),
+ drainInterval: TimeSpan.FromMilliseconds(20),
+ maxWriteAttempts: 2));
+
+ rec.Tell(new VirtualTagActor.DependencyValueChanged("Good.Tag", 1.0, DateTime.UtcNow));
+ rec.Tell(new VirtualTagActor.DependencyValueChanged("Bad.Tag", 2.0, DateTime.UtcNow));
+
+ await AwaitAssertAsync(async () =>
+ {
+ var status = await rec.Ask(
+ ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3));
+ Assert.Equal(0, status.QueuedDepth);
+ Assert.Equal(1, status.AbandonedEntries);
+ }, TimeSpan.FromSeconds(10));
+ }
+
+ ///
+ /// Scope pin: an raised while the drain token is NOT
+ /// cancelled is an ordinary write failure and still counts toward abandonment.
+ ///
+ ///
+ /// This is what the when (ct.IsCancellationRequested) filter buys: only a genuine shutdown
+ /// aborts the pass. A gateway that surfaces a cancellation-shaped exception from its own internal
+ /// deadline is a failing tag like any other — without the filter it would become an unquarantinable
+ /// tag that pins the FIFO head forever, re-opening #538 through the back door.
+ ///
+ [Fact]
+ public async Task An_OperationCanceledException_with_a_live_token_still_counts_toward_abandonment()
+ {
+ var mux = CreateTestProbe();
+ var writer = new FakeValueWriter { Succeed = true };
+ writer.SetThrowing("Bad.Tag", static _ => new OperationCanceledException("not our shutdown"));
+ var outbox = new InMemoryOutbox();
+
+ IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props(
+ mux.Ref, writer, outbox, historizedRefs: new[] { "Good.Tag", "Bad.Tag" },
+ minBackoff: TimeSpan.FromMilliseconds(20),
+ maxBackoff: TimeSpan.FromMilliseconds(50),
+ drainInterval: TimeSpan.FromMilliseconds(20),
+ maxWriteAttempts: 2));
+
+ rec.Tell(new VirtualTagActor.DependencyValueChanged("Good.Tag", 1.0, DateTime.UtcNow));
+ rec.Tell(new VirtualTagActor.DependencyValueChanged("Bad.Tag", 2.0, DateTime.UtcNow));
+
+ await AwaitAssertAsync(async () =>
+ {
+ var status = await rec.Ask(
+ ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3));
+ Assert.Equal(0, status.QueuedDepth);
+ Assert.Equal(1, status.AbandonedEntries);
+ }, TimeSpan.FromSeconds(10));
+ }
+
// ---------------- Gitea #537: only the recording Primary captures ----------------
///
@@ -1164,6 +1303,14 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
/// pass still succeed, and that contrast is the evidence the recorder keys on.
private readonly HashSet _failTags = new(StringComparer.Ordinal);
+ /// Historian tags whose writes THROW instead of returning an ack, keyed to the exception
+ /// factory to throw (given the drain's token). Distinct from because the
+ /// recorder's per-tag catch treats the two shapes differently: a returned false is always
+ /// a failure, whereas a thrown exception is a failure only when it is not the drain token's own
+ /// cancellation.
+ private readonly Dictionary> _throwTags =
+ new(StringComparer.Ordinal);
+
/// Starts or stops failing one tag. Callable mid-test so a case can prove a tag that
/// recovers within its budget is never abandoned.
/// The historian tag.
@@ -1177,6 +1324,18 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
}
}
+ /// Starts or stops THROWING for one tag's writes.
+ /// The historian tag.
+ /// Builds the exception to throw from the drain's token; null stops throwing.
+ public void SetThrowing(string tag, Func? factory)
+ {
+ lock (_gate)
+ {
+ if (factory is null) _throwTags.Remove(tag);
+ else _throwTags[tag] = factory;
+ }
+ }
+
/// Lifetime count of invocations (proves a retry ran).
public int CallCount
{
@@ -1208,6 +1367,13 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
_written.Add(new WrittenValue(tag, v.Value, v.Quality, v.TimestampUtc));
}
+ // A throwing tag overrides every ack rule. Thrown synchronously from inside the awaited
+ // call, which is where a real gateway client surfaces most faults.
+ if (_throwTags.TryGetValue(tag, out Func? factory))
+ {
+ throw factory(ct);
+ }
+
// Fail the first FailFirstN calls; thereafter honour Succeed — unless this specific tag
// is currently marked failing, which overrides both.
bool result = _calls > FailFirstN && Succeed && !_failTags.Contains(tag);
@@ -1216,6 +1382,90 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
}
}
+ ///
+ /// double for the shutdown-cancellation drain path: one
+ /// nominated tag returns a plain failed ack for its first failCalls writes and then BLOCKS
+ /// on the next one until the drain's token is cancelled, at which point it throws the
+ /// a real in-flight gateway call raises at node shutdown.
+ /// Every other tag succeeds immediately.
+ ///
+ ///
+ /// The block is what makes the cancellation land INSIDE a write rather than between passes — the
+ /// only place the per-tag catch can misclassify it. cannot express
+ /// this: it completes synchronously, so the drain token is never live-cancelled underneath it.
+ ///
+ private sealed class ShutdownBlockingValueWriter : IHistorianValueWriter
+ {
+ private readonly Lock _gate = new();
+ private readonly List _calls = new();
+ private readonly string _tag;
+ private readonly int _failCalls;
+ private readonly TaskCompletionSource _entered = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ private readonly TaskCompletionSource _cancelled = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ /// The historian tag that fails and then blocks.
+ /// How many of its writes return a plain failed ack before the blocking one.
+ public ShutdownBlockingValueWriter(string tag, int failCalls)
+ {
+ _tag = tag;
+ _failCalls = failCalls;
+ }
+
+ /// Completes once the blocking write has been ENTERED — i.e. the pass is mid-write and it
+ /// is safe to stop the recorder.
+ public Task Entered => _entered.Task;
+
+ /// Completes once that blocked write has observed the cancellation and thrown.
+ public Task Cancelled => _cancelled.Task;
+
+ /// Lifetime count of writes for one tag.
+ /// The historian tag.
+ /// How many times it has been written.
+ public int CallCountFor(string tag)
+ {
+ lock (_gate)
+ {
+ return _calls.Count(c => string.Equals(c, tag, StringComparison.Ordinal));
+ }
+ }
+
+ public async Task WriteLiveValuesAsync(
+ string tag, IReadOnlyList values, CancellationToken ct)
+ {
+ int seen;
+ lock (_gate)
+ {
+ _calls.Add(tag);
+ seen = _calls.Count(c => string.Equals(c, tag, StringComparison.Ordinal));
+ }
+
+ if (!string.Equals(tag, _tag, StringComparison.Ordinal))
+ {
+ return true;
+ }
+
+ if (seen <= _failCalls)
+ {
+ return false;
+ }
+
+ _entered.TrySetResult();
+ try
+ {
+ await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false);
+ }
+ catch (OperationCanceledException)
+ {
+ // Signalled before the rethrow so the test can await the exact moment the drain begins
+ // unwinding, rather than sleeping and hoping.
+ _cancelled.TrySetResult();
+ throw;
+ }
+
+ return false; // unreachable: the infinite delay only ever completes by cancelling
+ }
+ }
+
private sealed record WrittenValue(string Tag, double Value, ushort Quality, DateTime? TimestampUtc);
/// In-memory double honouring the FIFO-truncate