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

fix(runtime): stash dedup + applying-wedge watchdog; cache-failure diagnostics

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-15 17:43:15 -04:00
parent bfbdce32e4
commit 0dbbb5ac6b
3 changed files with 451 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorApplyingStashTests.cs
+276 -21
View File
@@ -3,7 +3,9 @@ using Akka.Actor;
using Akka.Event;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Interfaces;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
@@ -36,6 +38,10 @@ public sealed class DriverHostActorApplyingStashTests()
private static readonly NodeId TestNode = NodeId.Parse("driver-test");
private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64));
private static readonly RevisionHash RevB = RevisionHash.Parse(new string('b', 64));
private static readonly RevisionHash RevC = RevisionHash.Parse(new string('c', 64));
/// <summary>How long a forward-spin gets to announce itself thousands of times if it is back.</summary>
private static readonly TimeSpan SpinWindow = TimeSpan.FromSeconds(1.5);
/// <summary>Minimal parseable artifact — no drivers.</summary>
private static byte[] ArtifactBytes() =>
@@ -63,24 +69,20 @@ public sealed class DriverHostActorApplyingStashTests()
actor.Tell(new DispatchDeployment(deploymentA, RevA, CorrelationId.NewId()));
AwaitAssert(() => fetcher.Calls.ShouldBe(1), duration: PresenceBudget);
var logs = CreateTestProbe();
Sys.EventStream.Subscribe(logs.Ref, typeof(LogEvent));
var logs = SubscribeToLogs("deferring");
actor.Tell(new DispatchDeployment(deploymentB, RevB, CorrelationId.NewId()));
// Collect for long enough that a forward-spin would announce itself thousands of times.
var deferrals = 0;
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(1.5);
while (DateTime.UtcNow < deadline)
{
if (logs.ReceiveOne(TimeSpan.FromMilliseconds(150)) is LogEvent e &&
e.Message?.ToString()?.Contains("deferring", StringComparison.Ordinal) == true)
{
deferrals++;
}
}
// PRESENCE first, on the shared budget: the deferral must actually be ANNOUNCED. Counting
// only over a fixed window would let "the dispatch was silently dropped" score zero and
// then pass the ≤1 half below, which is the wrong bug to be blind to.
logs.DrainUntilAllSeen(PresenceBudget);
logs["deferring"].ShouldBeGreaterThanOrEqualTo(1);
deferrals.ShouldBe(1);
// ABSENCE second: over a window in which a forward-spin would announce itself thousands of
// times, no further deferral appears.
logs.Drain(SpinWindow);
logs["deferring"].ShouldBe(1);
// Nothing was lost: B re-enters exactly once, after A's apply transitions back to Steady.
fetcher.Release();
@@ -95,21 +97,274 @@ public sealed class DriverHostActorApplyingStashTests()
}
}
/// <summary>Gates the FIRST fetch on the test; later fetches complete immediately.</summary>
private sealed class GatedFetcher(byte[] bytes) : IDeploymentArtifactFetcher
/// <summary>
/// The SAME dispatch arriving twice mid-apply is stashed ONCE — the stash is bounded by
/// distinct deployments, not by how often the coordinator re-dispatches.
/// </summary>
/// <remarks>
/// <para>
/// Verified RED against the pinned Akka (1.5.62): with the dedup disabled, both copies are
/// stashed and the deferral is announced <b>twice</b>. That is the failure this pins — a
/// coordinator that re-dispatches on a timer would otherwise grow the stash with retries
/// rather than with work, and every extra copy is re-delivered on the way back to Steady
/// only to short-circuit as a duplicate.
/// </para>
/// <para>
/// It is NOT a crash today, and the review's suggestion that it might be was checked rather
/// than assumed. <c>AbstractStash.Stash()</c> is documented to throw
/// <c>IllegalActorStateException</c> when an envelope equals the last stashed one, and that
/// message string is in the shipped assembly — but a probe actor on this package stashed the
/// same instance twice, and two value-equal instances, with and without an explicit sender,
/// and never threw. The RED above is a count, not an exception, for exactly that reason.
/// </para>
/// </remarks>
[Fact]
public void TheSameDispatchArrivingTwiceMidApply_IsStashedOnce_AndAppliedOnce()
{
private readonly TaskCompletionSource _gate =
new(TaskCreationOptions.RunContinuationsAsynchronously);
var db = NewInMemoryDbFactory();
var fetcher = new GatedFetcher(ArtifactBytes());
var deploymentA = DeploymentId.NewId();
var deploymentB = DeploymentId.NewId();
var coordinator = CreateTestProbe();
var actor = Sys.ActorOf(DriverHostActor.Props(
db, TestNode, coordinator.Ref,
localRoles: new HashSet<string> { "driver" },
deploymentArtifactCache: new NullArtifactCache(),
fetchAndCacheMode: true,
artifactFetcher: fetcher));
try
{
actor.Tell(new DispatchDeployment(deploymentA, RevA, CorrelationId.NewId()));
AwaitAssert(() => fetcher.Calls.ShouldBe(1), duration: PresenceBudget);
var logs = SubscribeToLogs("deferring", "already deferred");
// The IDENTICAL message object, twice — a record, so the two envelopes are value-equal.
var dispatchB = new DispatchDeployment(deploymentB, RevB, CorrelationId.NewId());
actor.Tell(dispatchB);
actor.Tell(dispatchB);
logs.DrainUntilAllSeen(PresenceBudget);
logs["deferring"].ShouldBe(1);
logs["already deferred"].ShouldBe(1);
// No second stash, therefore no second re-delivery: exactly one Applied ack for B.
fetcher.Release();
coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(10)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(10)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
fetcher.Calls.ShouldBe(2);
}
finally
{
fetcher.Release();
}
}
/// <summary>
/// Two successive deferrals apply in arrival order, and the second one proves the dedup set is
/// CLEARED on the way back to Steady rather than accumulating across applies.
/// </summary>
/// <remarks>
/// C is stashed by B's apply, i.e. after A's apply already unstashed B and cleared the set. A
/// dedup set that was never cleared would still hold nothing for C, so what this really pins is
/// the re-stash CYCLE working at all — plus the ordering the coordinator depends on: the node
/// must end on C's revision, not on B's.
/// </remarks>
[Fact]
public void ASecondDeferralAppliesAfterTheFirst_AndTheNodeEndsOnTheLastRevision()
{
var db = NewInMemoryDbFactory();
// Gate the first TWO fetches so C can arrive while B is applying.
var fetcher = new GatedFetcher(ArtifactBytes(), gatedCalls: 2);
var deploymentA = DeploymentId.NewId();
var deploymentB = DeploymentId.NewId();
var deploymentC = DeploymentId.NewId();
var coordinator = CreateTestProbe();
var actor = Sys.ActorOf(DriverHostActor.Props(
db, TestNode, coordinator.Ref,
localRoles: new HashSet<string> { "driver" },
deploymentArtifactCache: new NullArtifactCache(),
fetchAndCacheMode: true,
artifactFetcher: fetcher));
try
{
actor.Tell(new DispatchDeployment(deploymentA, RevA, CorrelationId.NewId()));
AwaitAssert(() => fetcher.Calls.ShouldBe(1), duration: PresenceBudget);
actor.Tell(new DispatchDeployment(deploymentB, RevB, CorrelationId.NewId()));
// A completes; B unstashes and starts its own (still gated) fetch.
fetcher.Release(1);
coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(10)).DeploymentId.ShouldBe(deploymentA);
AwaitAssert(() => fetcher.Calls.ShouldBe(2), duration: PresenceBudget);
// C arrives mid-B — a fresh stash cycle, on a set that was cleared when B was released.
actor.Tell(new DispatchDeployment(deploymentC, RevC, CorrelationId.NewId()));
fetcher.Release(2);
coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(10)).DeploymentId.ShouldBe(deploymentB);
coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(10)).DeploymentId.ShouldBe(deploymentC);
actor.Tell(new GetDiagnostics(CorrelationId.NewId()), coordinator.Ref);
coordinator.ExpectMsg<NodeDiagnosticsSnapshot>(TimeSpan.FromSeconds(5))
.CurrentRevision.ShouldBe(RevC);
}
finally
{
fetcher.Release();
}
}
/// <summary>
/// A host wedged in <c>Applying</c> says so. The pre-stash forward-spin was accidentally the
/// only alarm for this — a flood of log lines nobody could miss — and removing it removed the
/// signal with it. The watchdog is the deliberate replacement.
/// </summary>
/// <remarks>
/// The wedge delay is a constructor parameter (defaulting to
/// <see cref="DriverHostActor.DefaultApplyWedgeTimeout"/>), shortened here so the fire is
/// observable in a test's lifetime — the same seam every other interval on this actor uses.
/// Nothing is cancelled or failed by the fire, so the assertion is purely the Warning, and it
/// must NAME the wedged deployment or it is useless during an incident.
/// </remarks>
[Fact]
public void AnApplyThatNeverCompletes_IsReportedAsWedged()
{
var db = NewInMemoryDbFactory();
var fetcher = new GatedFetcher(ArtifactBytes());
var deploymentA = DeploymentId.NewId();
var coordinator = CreateTestProbe();
var actor = Sys.ActorOf(DriverHostActor.Props(
db, TestNode, coordinator.Ref,
localRoles: new HashSet<string> { "driver" },
deploymentArtifactCache: new NullArtifactCache(),
fetchAndCacheMode: true,
artifactFetcher: fetcher,
applyWedgeTimeout: TimeSpan.FromMilliseconds(300)));
try
{
EventFilter.Warning(contains: deploymentA.ToString()).ExpectOne(TimeSpan.FromSeconds(10), () =>
actor.Tell(new DispatchDeployment(deploymentA, RevA, CorrelationId.NewId())));
}
finally
{
fetcher.Release();
}
}
/// <summary>
/// A completed apply does NOT report a wedge, however short the delay — the timer is cancelled
/// on the way back to Steady, and a fire already in the mailbox drops at Debug.
/// </summary>
/// <remarks>
/// The falsification control for the test above: a watchdog that fires unconditionally would
/// pass it and would then cry wedge after every ordinary deployment.
/// </remarks>
[Fact]
public void ACompletedApply_NeverReportsAWedge()
{
var db = NewInMemoryDbFactory();
var deploymentA = DeploymentId.NewId();
var coordinator = CreateTestProbe();
var actor = Sys.ActorOf(DriverHostActor.Props(
db, TestNode, coordinator.Ref,
localRoles: new HashSet<string> { "driver" },
deploymentArtifactCache: new NullArtifactCache(),
fetchAndCacheMode: true,
// Ungated: the fetch answers immediately.
artifactFetcher: new GatedFetcher(ArtifactBytes(), gatedCalls: 0),
applyWedgeTimeout: TimeSpan.FromMilliseconds(200)));
var logs = SubscribeToLogs("appears WEDGED");
actor.Tell(new DispatchDeployment(deploymentA, RevA, CorrelationId.NewId()));
coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(10)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
// Well past the (200 ms) delay — a live timer would have fired several times over.
logs.Drain(SpinWindow);
logs["appears WEDGED"].ShouldBe(0);
}
/// <summary>Subscribes a fresh counter to the EventStream's log events.</summary>
/// <remarks>
/// Every fragment of interest must be declared up front, because draining is destructive: a
/// probe read for one fragment discards every other log event it passes over, so counting two
/// fragments with two independent drains would always score zero on the second.
/// </remarks>
private LogCounter SubscribeToLogs(params string[] fragments)
{
var probe = CreateTestProbe();
Sys.EventStream.Subscribe(probe.Ref, typeof(LogEvent));
return new LogCounter(probe, fragments);
}
/// <summary>Counts occurrences of a fixed set of log-message fragments off one EventStream probe.</summary>
private sealed class LogCounter(Akka.TestKit.TestProbe logs, string[] fragments)
{
private readonly Dictionary<string, int> _counts =
fragments.ToDictionary(f => f, _ => 0, StringComparer.Ordinal);
public int this[string fragment] => _counts[fragment];
/// <summary>Drains the probe for a fixed window, tallying every tracked fragment.</summary>
public void Drain(TimeSpan window)
{
var deadline = DateTime.UtcNow + window;
while (DateTime.UtcNow < deadline)
{
if (logs.ReceiveOne(TimeSpan.FromMilliseconds(100)) is LogEvent e &&
e.Message?.ToString() is { } text)
{
foreach (var fragment in fragments)
if (text.Contains(fragment, StringComparison.Ordinal))
_counts[fragment]++;
}
}
}
/// <summary>Drains until every tracked fragment has been seen at least once, or the budget expires.</summary>
public void DrainUntilAllSeen(TimeSpan budget)
{
var deadline = DateTime.UtcNow + budget;
while (DateTime.UtcNow < deadline && _counts.Values.Any(n => n == 0))
Drain(TimeSpan.FromMilliseconds(200));
}
}
/// <summary>
/// Gates the first <c>gatedCalls</c> fetches on the test; later fetches complete immediately.
/// </summary>
private sealed class GatedFetcher(byte[] bytes, int gatedCalls = 1) : IDeploymentArtifactFetcher
{
private readonly TaskCompletionSource[] _gates = Enumerable.Range(0, gatedCalls)
.Select(_ => new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously))
.ToArray();
private int _calls;
public int Calls => Volatile.Read(ref _calls);
public void Release() => _gate.TrySetResult();
/// <summary>Releases every gate — also the test-teardown safety net.</summary>
public void Release()
{
foreach (var gate in _gates) gate.TrySetResult();
}
/// <summary>Releases the gate on the <paramref name="call"/>-th fetch (1-based).</summary>
public void Release(int call) => _gates[call - 1].TrySetResult();
public async Task<byte[]?> FetchAsync(string deploymentId, string expectedRevisionHash, CancellationToken ct)
{
if (Interlocked.Increment(ref _calls) == 1)
await _gate.Task.ConfigureAwait(false);
var call = Interlocked.Increment(ref _calls);
if (call <= _gates.Length)
await _gates[call - 1].Task.ConfigureAwait(false);
return bytes;
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorAsyncCacheWriteTests.cs
+15 -5
View File
@@ -1,5 +1,6 @@
using System.Text.Json;
using Akka.Actor;
using Akka.Event;
using Microsoft.EntityFrameworkCore;
using Shouldly;
using Xunit;
@@ -93,11 +94,20 @@ public sealed class DriverHostActorAsyncCacheWriteTests()
localRoles: new HashSet<string> { "driver" },
deploymentArtifactCache: new AsyncFailingArtifactCache()));
EventFilter.Warning(contains: "could not be cached").ExpectOne(TimeSpan.FromSeconds(5), () =>
{
actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
});
// Matched on the EXCEPTION, not just the prose: the pipe-back carries the Exception itself and
// the handler logs it as the event's Cause. A SQLite failure is routinely a wrapped inner
// exception whose outer Message names neither the result code nor the failing frame, so a
// Reason-string-only carrier would have thrown that away before the log line was written.
// (Custom rather than EventFilter.Exception<T>, which only ever matches ERROR-level events —
// a failed cache write is deliberately a Warning, since the apply itself succeeded.)
EventFilter.Custom<Warning>(e =>
e.Cause is InvalidOperationException &&
e.Message?.ToString()?.Contains("could not be cached", StringComparison.Ordinal) == true)
.ExpectOne(TimeSpan.FromSeconds(5), () =>
{
actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
});
coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
}