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

fix(historian): a shutdown cancellation mid-drain is a pass abort, not a per-tag failure

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-17 03:52:55 -04:00
parent dba5f806d3
commit 7870c91fe4
2 changed files with 264 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs
+250
View File
@@ -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) -------
/// <summary>
/// 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.
/// </summary>
/// <remarks>
/// The per-tag lambda wrapped the writer call in a bare <c>catch</c>, so a shutdown
/// <see cref="OperationCanceledException"/> 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
/// <c>catch (OperationCanceledException) when (ct.IsCancellationRequested)</c> already treats a
/// cancelled pass as a non-fault, but the inner catch ate the exception before it could be reached.
/// <para>
/// 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.
/// </para>
/// <para>
/// The budget is deliberately the tightest there is (<c>maxWriteAttempts: 1</c>) 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.
/// </para>
/// </remarks>
[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));
}
/// <summary>
/// Scope pin: a NON-cancellation throw from one tag, while a sibling wrote, still counts toward
/// the Gitea #538 budget and still abandons.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
[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.RecorderStatus>(
ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3));
Assert.Equal(0, status.QueuedDepth);
Assert.Equal(1, status.AbandonedEntries);
}, TimeSpan.FromSeconds(10));
}
/// <summary>
/// Scope pin: an <see cref="OperationCanceledException"/> raised while the drain token is NOT
/// cancelled is an ordinary write failure and still counts toward abandonment.
/// </summary>
/// <remarks>
/// This is what the <c>when (ct.IsCancellationRequested)</c> 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.
/// </remarks>
[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.RecorderStatus>(
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 ----------------
/// <summary>
@@ -1164,6 +1303,14 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
/// pass still succeed, and that contrast is the evidence the recorder keys on.</summary>
private readonly HashSet<string> _failTags = new(StringComparer.Ordinal);
/// <summary>Historian tags whose writes THROW instead of returning an ack, keyed to the exception
/// factory to throw (given the drain's token). Distinct from <see cref="_failTags"/> because the
/// recorder's per-tag catch treats the two shapes differently: a returned <c>false</c> is always
/// a failure, whereas a thrown exception is a failure only when it is not the drain token's own
/// cancellation.</summary>
private readonly Dictionary<string, Func<CancellationToken, Exception>> _throwTags =
new(StringComparer.Ordinal);
/// <summary>Starts or stops failing one tag. Callable mid-test so a case can prove a tag that
/// recovers within its budget is never abandoned.</summary>
/// <param name="tag">The historian tag.</param>
@@ -1177,6 +1324,18 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
}
}
/// <summary>Starts or stops THROWING for one tag's writes.</summary>
/// <param name="tag">The historian tag.</param>
/// <param name="factory">Builds the exception to throw from the drain's token; null stops throwing.</param>
public void SetThrowing(string tag, Func<CancellationToken, Exception>? factory)
{
lock (_gate)
{
if (factory is null) _throwTags.Remove(tag);
else _throwTags[tag] = factory;
}
}
/// <summary>Lifetime count of <see cref="WriteLiveValuesAsync"/> invocations (proves a retry ran).</summary>
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<CancellationToken, Exception>? 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
}
}
/// <summary>
/// <see cref="IHistorianValueWriter"/> double for the shutdown-cancellation drain path: one
/// nominated tag returns a plain failed ack for its first <c>failCalls</c> writes and then BLOCKS
/// on the next one until the drain's token is cancelled, at which point it throws the
/// <see cref="OperationCanceledException"/> a real in-flight gateway call raises at node shutdown.
/// Every other tag succeeds immediately.
/// </summary>
/// <remarks>
/// 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. <see cref="FakeValueWriter"/> cannot express
/// this: it completes synchronously, so the drain token is never live-cancelled underneath it.
/// </remarks>
private sealed class ShutdownBlockingValueWriter : IHistorianValueWriter
{
private readonly Lock _gate = new();
private readonly List<string> _calls = new();
private readonly string _tag;
private readonly int _failCalls;
private readonly TaskCompletionSource _entered = new(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly TaskCompletionSource _cancelled = new(TaskCreationOptions.RunContinuationsAsynchronously);
/// <param name="tag">The historian tag that fails and then blocks.</param>
/// <param name="failCalls">How many of its writes return a plain failed ack before the blocking one.</param>
public ShutdownBlockingValueWriter(string tag, int failCalls)
{
_tag = tag;
_failCalls = failCalls;
}
/// <summary>Completes once the blocking write has been ENTERED — i.e. the pass is mid-write and it
/// is safe to stop the recorder.</summary>
public Task Entered => _entered.Task;
/// <summary>Completes once that blocked write has observed the cancellation and thrown.</summary>
public Task Cancelled => _cancelled.Task;
/// <summary>Lifetime count of writes for one tag.</summary>
/// <param name="tag">The historian tag.</param>
/// <returns>How many times it has been written.</returns>
public int CallCountFor(string tag)
{
lock (_gate)
{
return _calls.Count(c => string.Equals(c, tag, StringComparison.Ordinal));
}
}
public async Task<bool> WriteLiveValuesAsync(
string tag, IReadOnlyList<HistorizationValue> 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);
/// <summary>In-memory <see cref="IHistorizationOutbox"/> double honouring the FIFO-truncate