test(runtime): operator restart re-subscribes the respawned child (archreview 03/S16, U6)
Seen RED against a T3-reverted build: all 4 failed, Created[1].SubscribeCount == 0 on the headline case. 20x isolation run green after the fix was restored.
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,287 @@
|
||||
using Akka.Actor;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Types;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
|
||||
|
||||
/// <summary>
|
||||
/// The repo's first driver-child <b>restart</b> suite (archreview 03/S16, 03/U6): what a child obtained
|
||||
/// by each of the three incarnation routes is actually handed.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>The falsification bar, and why this file exists at all.</b> An actor-restart test must assert a
|
||||
/// FUNCTION the new incarnation performs — it subscribed, it re-initialised — never that the
|
||||
/// incarnation EXISTS. Existence is exactly what
|
||||
/// <c>DriverHostActorReconcileTests.RestartDriver_respawns_the_child_without_an_actor_name_collision</c>
|
||||
/// asserts (no name collision, factory created a second driver, one child remains), and that test
|
||||
/// passes green against the 03/S16 defect: a respawned child that connects and then reads nothing
|
||||
/// forever. It is kept, because its own subject — the <c>InvalidActorNameException</c> — is real; it
|
||||
/// is simply the weaker half of a pair, and this file is the other half.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>The three incarnation routes, and what re-provisions each.</b> (1) <i>Deploy reconcile</i> —
|
||||
/// <c>ApplyDriverSpecs</c> → <c>SpawnChild</c>, then the apply's bulk desired-subscription push.
|
||||
/// (2) <i>Operator Restart</i> — <c>HandleRestartDriver</c> stops the child and calls
|
||||
/// <c>SpawnChild</c> with the cached spec; before 03/S16 that was the whole story, and the new child
|
||||
/// started with an EMPTY desired set that nothing re-sent until the next deployment (the #488
|
||||
/// periodic reconcile early-returns on an empty set, so it is structurally inert here).
|
||||
/// (3) <i>Supervision restart</i> — Akka's default decider recreates the actor object around the
|
||||
/// SAME driver instance without going through <c>SpawnChild</c> at all, so nothing re-sent
|
||||
/// <c>InitializeRequested</c> either and the child sat in <c>Connecting</c> forever around a driver
|
||||
/// its own <c>PostStop</c> had already shut down. A fourth route, <i>Force Reconnect</i>, re-uses the
|
||||
/// live actor instance and needs nothing — a verified negative, pinned here so a future edit to the
|
||||
/// restart path cannot break the path that was already correct.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>The supervision crash seam, and why the obvious ones do not work.</b> The seam used is the
|
||||
/// driver's own <c>OnDataChange</c> carrying a NULL <c>Snapshot</c>
|
||||
/// (<see cref="RecordingSubscribableDriver.FireMalformedDataChange"/>): <c>DataChangeEventArgs</c> is
|
||||
/// a positional record with no null validation, and <c>DriverInstanceActor.OnDataChangeForward</c>
|
||||
/// dereferences <c>msg.Snapshot</c> unguarded on the ACTOR thread, so it NREs inside the child and
|
||||
/// the default decider Restarts it. Rejected alternatives, each for a specific reason: a throwing
|
||||
/// <c>IDriverHealthPublisher</c> (<c>PublishHealthSnapshot</c> wraps the publish in a
|
||||
/// <c>catch (Exception)</c>, so it never escapes); a throwing <c>SubscribeAsync</c> (also swallowed —
|
||||
/// the stub already models it via <c>SubscribeFailuresBeforeSuccess</c>); <c>Kill.Instance</c> (the
|
||||
/// default decider <b>Stops</b> on <c>ActorKilledException</c>, a different directive and therefore a
|
||||
/// different code path); and an <c>internal</c> "throw this" test message (a production surface added
|
||||
/// purely for a test). The fault is arbitrary by design — what is under test is the RECOVERY. If
|
||||
/// <c>DataChangeEventArgs</c> ever gains null validation, the fallback is an explicit
|
||||
/// <c>OneForOneStrategy(Directive.Restart)</c> on the host plus an internal test-only throw message.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class DriverHostActorRestartTests : RuntimeActorTestBase
|
||||
{
|
||||
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 NodeId TestNode = NodeId.Parse("node-1:4053");
|
||||
|
||||
private const string DriverId = "DI-1";
|
||||
private const string SpeedPath = "Plant/Modbus/dev1/speed";
|
||||
private const string TempPath = "Plant/Modbus/dev1/temp";
|
||||
|
||||
/// <summary>Seeds + applies a v3 deployment whose single Modbus driver is ENABLED, so a real child is
|
||||
/// spawned around a <see cref="RecordingSubscribableDriver"/>, and waits until that child has connected
|
||||
/// and established its first subscription.</summary>
|
||||
private (IActorRef Host, RecordingSubscribableDriverFactory Factory) DeployOneDriver(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext> db,
|
||||
(string Driver, string DriverName, string Device, string Tag, string DataType, bool Writable)[] rawTags,
|
||||
IReadOnlyDictionary<string, string>? tagConfigByTag = null)
|
||||
{
|
||||
var factory = new RecordingSubscribableDriverFactory("Modbus");
|
||||
var deploymentId = V3ArtifactSeeder.SeedV3Deployment(
|
||||
db, RevA, rawTags, Array.Empty<(string, string, string, string, string?)>(),
|
||||
enabled: true, tagConfigByTag: tagConfigByTag);
|
||||
var (host, _, _) = V3ArtifactSeeder.SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, factory);
|
||||
return (host, factory);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <b>The 03/S16 falsification.</b> An operator's Restart is supposed to recover a sick driver; before
|
||||
/// the fix it silently killed it — the respawned child came up with <c>_desiredRefs</c> empty, and the
|
||||
/// only <c>SetDesiredSubscriptions</c> sender in the repo was the apply's bulk push, so the driver
|
||||
/// subscribed to nothing until the next deployment dispatch.
|
||||
/// <para>
|
||||
/// Deliberately NOT the pass criterion: that a second driver was created, or that one child
|
||||
/// remains. Both are true against the defect.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Operator_restart_re_subscribes_the_respawned_child()
|
||||
{
|
||||
var db = NewInMemoryDbFactory();
|
||||
var (host, factory) = DeployOneDriver(db, new[] { (DriverId, "Modbus", "dev1", "speed", "Float", false) });
|
||||
|
||||
AwaitAssert(
|
||||
() =>
|
||||
{
|
||||
factory.Created.Count.ShouldBe(1);
|
||||
factory.Created[0].SubscribeCount.ShouldBe(1);
|
||||
factory.Created[0].LastSubscribedRefs.ShouldBe(new[] { SpeedPath });
|
||||
},
|
||||
duration: PresenceBudget);
|
||||
|
||||
host.Tell(new RestartDriver("c1", DriverId, "tester", Guid.NewGuid()));
|
||||
|
||||
AwaitAssert(
|
||||
() =>
|
||||
{
|
||||
factory.Created.Count.ShouldBe(2, "the restart did not respawn the child around a new driver");
|
||||
factory.Created[1].SubscribeCount.ShouldBeGreaterThanOrEqualTo(
|
||||
1,
|
||||
"the RESPAWNED child never subscribed — it is Connected and reading nothing, which is the " +
|
||||
"03/S16 defect (an operator pressed Restart to recover a driver and silently killed it)");
|
||||
factory.Created[1].LastSubscribedRefs.ShouldBe(new[] { SpeedPath });
|
||||
},
|
||||
duration: PresenceBudget);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The value refs and the alarm refs ride in ONE <c>SetDesiredSubscriptions</c> but are applied by two
|
||||
/// different child code paths (the value set self-Tells <c>Subscribe</c>; the alarm set runs
|
||||
/// <c>SubscribeDesiredAlarms</c> against the <c>IAlarmSource</c> feed, which stays gated shut until it
|
||||
/// is called). Sending the message is therefore not evidence that both halves landed, so the alarm
|
||||
/// half is pinned separately.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Operator_restart_re_subscribes_alarm_refs_too()
|
||||
{
|
||||
var db = NewInMemoryDbFactory();
|
||||
var (host, factory) = DeployOneDriver(
|
||||
db,
|
||||
new[]
|
||||
{
|
||||
(DriverId, "Modbus", "dev1", "speed", "Float", false),
|
||||
(DriverId, "Modbus", "dev1", "temp", "Float", false),
|
||||
},
|
||||
// An alarm-bearing raw tag is EXCLUDED from the value set and carried in the alarm set instead.
|
||||
new Dictionary<string, string>
|
||||
{
|
||||
["temp"] = """{"alarm":{"alarmType":"AlarmCondition","severity":500}}""",
|
||||
});
|
||||
|
||||
AwaitAssert(
|
||||
() =>
|
||||
{
|
||||
factory.Created.Count.ShouldBe(1);
|
||||
factory.Created[0].SubscribeAlarmsCount.ShouldBeGreaterThanOrEqualTo(1);
|
||||
factory.Created[0].LastSubscribedAlarmRefs.ShouldBe(new[] { TempPath });
|
||||
},
|
||||
duration: PresenceBudget);
|
||||
|
||||
host.Tell(new RestartDriver("c1", DriverId, "tester", Guid.NewGuid()));
|
||||
|
||||
AwaitAssert(
|
||||
() =>
|
||||
{
|
||||
factory.Created.Count.ShouldBe(2);
|
||||
factory.Created[1].SubscribeAlarmsCount.ShouldBeGreaterThanOrEqualTo(
|
||||
1,
|
||||
"the respawned child never re-subscribed its alarm refs — the IAlarmSource feed stays " +
|
||||
"gated shut, so no native alarm transition can ever reach the host again");
|
||||
factory.Created[1].LastSubscribedAlarmRefs.ShouldBe(new[] { TempPath });
|
||||
// The value half must survive the alarm half being present.
|
||||
factory.Created[1].LastSubscribedRefs.ShouldBe(new[] { SpeedPath });
|
||||
},
|
||||
duration: PresenceBudget);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The ordering guard for the spawn-time send. <c>ApplyDriverSpecs</c> (which spawns) runs BEFORE
|
||||
/// <c>ApplyDesiredSubscriptions</c> (which rebuilds the authority), so a child spawned by a fresh
|
||||
/// deploy is momentarily handed the PREVIOUS apply's set. That transient must never be acted on: it
|
||||
/// is superseded by the bulk push on the same sender→receiver pair (Akka guarantees per-pair
|
||||
/// ordering) while the child is still <c>Connecting</c>, where <c>SetDesiredSubscriptions</c> only
|
||||
/// STORES; the subscribe happens on the Connected entry, from whatever was stored last.
|
||||
/// <para>
|
||||
/// Revision B swaps the driver's tag, so A's and B's sets are distinguishable. Note the raw tags
|
||||
/// are folded into the merged <c>DriverConfig</c>, so a tag-set change DOES change the driver
|
||||
/// spec and therefore respawns the child (#516) — which is precisely the arrangement under test:
|
||||
/// a brand-new child, spawned while the authority still holds A, must end up subscribed to B.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>What is asserted is CONVERGENCE, not inertness</b>, and the difference was measured rather
|
||||
/// than assumed. The child runs concurrently with the rest of the apply handler, so a driver
|
||||
/// whose <c>InitializeAsync</c> completes instantly (this stub does) can reach Connected and
|
||||
/// subscribe the transient stale set BEFORE the bulk push arrives — observed here as
|
||||
/// <c>SubscribeCount == 2</c>. Akka's per-pair ordering still guarantees the final state is B's,
|
||||
/// and <c>HandleSubscribeAsync</c> drops the prior subscription before establishing the new one,
|
||||
/// so the cost is one extra round-trip on a respawning deploy. Pinning the count would pin a
|
||||
/// race; the final refs are the invariant.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void A_child_spawned_by_a_fresh_deploy_ends_with_the_NEW_refs()
|
||||
{
|
||||
var db = NewInMemoryDbFactory();
|
||||
var (host, factory) = DeployOneDriver(db, new[] { (DriverId, "Modbus", "dev1", "speed", "Float", false) });
|
||||
|
||||
AwaitAssert(
|
||||
() => factory.Created[0].LastSubscribedRefs.ShouldBe(new[] { SpeedPath }),
|
||||
duration: PresenceBudget);
|
||||
|
||||
var revB = V3ArtifactSeeder.SeedV3Deployment(
|
||||
db, RevB,
|
||||
new[] { (DriverId, "Modbus", "dev1", "temp", "Float", false) },
|
||||
Array.Empty<(string, string, string, string, string?)>(),
|
||||
enabled: true);
|
||||
host.Tell(new Commons.Messages.Deploy.DispatchDeployment(revB, RevB, CorrelationId.NewId()));
|
||||
|
||||
AwaitAssert(
|
||||
() =>
|
||||
{
|
||||
factory.Created.Count.ShouldBe(2, "the tag-set change re-spawns the child (#516 config respawn)");
|
||||
factory.Created[1].LastSubscribedRefs.ShouldBe(
|
||||
new[] { TempPath },
|
||||
"the newly-spawned child is on the PREVIOUS revision's refs — the spawn-time set was not superseded");
|
||||
},
|
||||
duration: PresenceBudget);
|
||||
|
||||
// A Restart on top of that must hand the new child B's set, not A's — i.e. the authority really is
|
||||
// repopulated WHOLE on every apply rather than accumulated.
|
||||
host.Tell(new RestartDriver("c1", DriverId, "tester", Guid.NewGuid()));
|
||||
AwaitAssert(
|
||||
() =>
|
||||
{
|
||||
factory.Created.Count.ShouldBe(3);
|
||||
factory.Created[2].LastSubscribedRefs.ShouldBe(new[] { TempPath });
|
||||
},
|
||||
duration: PresenceBudget);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The empty-set semantics are load-bearing and must not degrade into "skip the send": an EMPTY value
|
||||
/// set is what tells the child to <c>Unsubscribe</c> and drop a stale handle, so it must be SENT.
|
||||
/// <para>
|
||||
/// The subject is a driver whose only raw tag is alarm-bearing: alarm tags are excluded from the
|
||||
/// value set and carried in the alarm one, so its value half is genuinely empty while its alarm
|
||||
/// half is not — and both halves ride in ONE <c>SetDesiredSubscriptions</c>. That is what makes
|
||||
/// the absence assertion non-vacuous: <c>SubscribeAlarmsCount >= 1</c> proves the message
|
||||
/// ARRIVED, and <c>SubscribeCount == 0</c> then proves the empty value half was honoured rather
|
||||
/// than the whole send being skipped. Asserting only "it did not subscribe" would pass equally
|
||||
/// against a build that sent the respawned child nothing at all — the 03/S16 defect itself.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Restart_of_a_driver_with_an_empty_value_set_still_sends_the_set()
|
||||
{
|
||||
var db = NewInMemoryDbFactory();
|
||||
var (host, factory) = DeployOneDriver(
|
||||
db,
|
||||
new[] { (DriverId, "Modbus", "dev1", "temp", "Float", false) },
|
||||
new Dictionary<string, string>
|
||||
{
|
||||
["temp"] = """{"alarm":{"alarmType":"AlarmCondition","severity":500}}""",
|
||||
});
|
||||
|
||||
AwaitAssert(
|
||||
() =>
|
||||
{
|
||||
factory.Created.Count.ShouldBe(1);
|
||||
factory.Created[0].SubscribeAlarmsCount.ShouldBeGreaterThanOrEqualTo(1);
|
||||
},
|
||||
duration: PresenceBudget);
|
||||
|
||||
host.Tell(new RestartDriver("c1", DriverId, "tester", Guid.NewGuid()));
|
||||
|
||||
AwaitAssert(
|
||||
() =>
|
||||
{
|
||||
factory.Created.Count.ShouldBe(2);
|
||||
var respawned = factory.Created[1];
|
||||
respawned.InitializeCount.ShouldBe(1, "the respawned child was never initialised");
|
||||
// Positive control: the SetDesiredSubscriptions genuinely reached the respawned child.
|
||||
respawned.SubscribeAlarmsCount.ShouldBeGreaterThanOrEqualTo(
|
||||
1, "the respawned child received no desired set at all — the absence below would be vacuous");
|
||||
respawned.LastSubscribedAlarmRefs.ShouldBe(new[] { TempPath });
|
||||
respawned.SubscribeCount.ShouldBe(
|
||||
0, "an EMPTY value set must route to Unsubscribe / no-op, never to a spurious subscribe");
|
||||
},
|
||||
duration: PresenceBudget);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user