From 2b7100cf19ae1a7cb33d3a4b757a3a38ae826870 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 20 Aug 2026 10:27:02 -0400 Subject: [PATCH] fix(runtime): route a reprovision by incarnation, not driver id (review F2) A ReprovisionRequested in flight from an old incarnation could reach the NEW child after HandleRestartDriver swapped it in. In Connecting that child DOES handle InitializeRequested, so it started a second concurrent InitializeAsync, racing the #527 dispose-before-overwrite. Seen RED: InitializeCount == 2 on a child held in Connecting by a gated init (an already-connected child swallows InitializeRequested and passes vacuously). --- .../Drivers/DriverHostActor.cs | 14 +++++ .../Drivers/DriverHostActorRestartTests.cs | 59 +++++++++++++++++++ .../Drivers/StubDrivers.cs | 25 +++++++- 3 files changed, 95 insertions(+), 3 deletions(-) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 00352e15..588e40ee 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -3017,6 +3017,20 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // Already stopped (a restart racing a reconcile that removed the driver) — nothing to provision. if (!_children.TryGetValue(msg.DriverInstanceId, out var entry)) return; + // Route by INCARNATION, not just by id. This message is delivered asynchronously, so one can still + // be in flight from an OLD incarnation after HandleRestartDriver stopped that child and spawned a + // replacement under the same DriverInstanceId. Delivered to the new child — which is in Connecting, + // where InitializeRequested IS handled — it would start a second, concurrent InitializeAsync, + // racing the dispose-before-overwrite that #527's teardown exists to serialise + // (ModbusDriver._transport, MqttDriver._connection each leak a transport on that race). + if (!Equals(Sender, entry.Actor)) + { + _log.Debug( + "DriverHost {Node}: dropping a reprovision for {Id} from a superseded incarnation ({Stale})", + _localNode, msg.DriverInstanceId, Sender); + return; + } + // A crash that Akka silently restarted is otherwise invisible to operators; name it. _log.Warning( "DriverHost {Node}: driver child {Id} was restarted by supervision after {Reason} — re-provisioning", diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorRestartTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorRestartTests.cs index 020e36d4..53652f9f 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorRestartTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorRestartTests.cs @@ -5,6 +5,7 @@ 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.Drivers; using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; @@ -419,6 +420,64 @@ public sealed class DriverHostActorRestartTests : RuntimeActorTestBase duration: PresenceBudget); } + /// + /// Review finding F2 — a reprovision must be routed by INCARNATION, not just by driver id. + /// ReprovisionRequested is delivered asynchronously, so one can still be in flight from an old + /// incarnation when HandleRestartDriver has already stopped that child and swapped in a new + /// one under the same DriverInstanceId. Routing on the id alone delivers it to the NEW child, + /// which is in Connecting around a healthy driver: it triggers a second, concurrent + /// InitializeAsync that races the very dispose-before-overwrite the #527 teardown exists to + /// serialise (ModbusDriver._transport, MqttDriver._connection). + /// + /// A stale sender is simulated directly — a probe standing in for the dead incarnation — because + /// provoking the real race would be timing-dependent. The assertion is an ABSENCE, so it carries + /// a positive control: the current child is proven alive and provisioned first, and its + /// InitializeCount must then not move. + /// + /// + [Fact] + public void A_reprovision_from_a_stale_incarnation_is_ignored() + { + var db = NewInMemoryDbFactory(); + // The child must be in CONNECTING for this defect to bite: InitializeRequested is unhandled in the + // Connected behaviour, so an already-connected child would swallow the stale reprovision for the + // wrong reason and the test would pass vacuously. The gated init holds it in Connecting. + var factory = new RecordingSubscribableDriverFactory("Modbus", gateInitialize: true); + var deploymentId = V3ArtifactSeeder.SeedV3Deployment( + db, RevA, + new[] { (DriverId, "Modbus", "dev1", "speed", "Float", false) }, + Array.Empty<(string, string, string, string, string?)>(), + enabled: true); + var (host, _, _) = V3ArtifactSeeder.SpawnHostAndApply(this, db, TestNode, deploymentId, RevA, factory); + + // Positive control: the current child is alive and has begun its one and only initialise. + AwaitAssert( + () => + { + factory.Created.Count.ShouldBe(1); + factory.Created[0].InitializeCount.ShouldBe(1); + }, + duration: PresenceBudget); + var driver = factory.Created[0]; + + // A ReprovisionRequested that did NOT come from the child the host currently holds. + var staleIncarnation = CreateTestProbe(); + host.Tell( + new DriverInstanceActor.ReprovisionRequested(DriverId, "NullReferenceException"), + staleIncarnation.Ref); + + // Round-trip a message the host must answer, so the reprovision above is provably processed before + // the absence is asserted — otherwise "nothing happened yet" would pass for "it was ignored". + var coordinator = CreateTestProbe(); + host.Tell(new Commons.Messages.Fleet.GetDiagnostics(CorrelationId.NewId()), coordinator.Ref); + coordinator.ExpectMsg(PresenceBudget); + + driver.InitializeCount.ShouldBe( + 1, + "a reprovision from a dead incarnation reached the CURRENT child and started a second, " + + "concurrent InitializeAsync on a driver that was already connected"); + } + /// /// The verified negative, pinned so this plan's edits to the restart paths cannot break the one that /// was already correct. Force Reconnect (ReconnectDriver) Tells ForceReconnect to the diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs index 30a51113..3c4783d0 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs @@ -261,11 +261,19 @@ internal sealed class RecordingSubscribableDriver : IDriver, ISubscribable, IAla get { lock (_gate) return _lastSubscribedAlarmRefs; } } + /// + /// When set, awaits it instead of completing inline — so the actor + /// stays in Connecting for as long as the test wants. Needed because several defects only + /// bite a child that has NOT yet connected (InitializeRequested is unhandled in + /// Connected, so a test built on an already-connected child passes vacuously). + /// + public TaskCompletionSource? InitializeGate { get; set; } + /// - public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) + public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { Interlocked.Increment(ref _initializeCount); - return Task.CompletedTask; + if (InitializeGate is { } gate) await gate.Task.ConfigureAwait(false); } /// @@ -359,9 +367,18 @@ internal sealed class RecordingSubscribableDriverFactory : IDriverFactory private readonly Lock _gate = new(); private readonly List _created = new(); + private readonly bool _gateInitialize; + /// Initializes a factory that materialises exactly one driver type. /// The driver type this factory answers for. - public RecordingSubscribableDriverFactory(string supportedType) => _supportedType = supportedType; + /// When true, every created driver gets an un-completed + /// , so its child actor stays in + /// Connecting until the test completes it. + public RecordingSubscribableDriverFactory(string supportedType, bool gateInitialize = false) + { + _supportedType = supportedType; + _gateInitialize = gateInitialize; + } /// Every driver this factory created, in creation order. public IReadOnlyList Created @@ -374,6 +391,8 @@ internal sealed class RecordingSubscribableDriverFactory : IDriverFactory { if (!string.Equals(driverType, _supportedType, StringComparison.Ordinal)) return null; var driver = new RecordingSubscribableDriver(driverInstanceId, driverType); + if (_gateInitialize) + driver.InitializeGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); lock (_gate) _created.Add(driver); return driver; }