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).
This commit is contained in:
@@ -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",
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <b>Review finding F2 — a reprovision must be routed by INCARNATION, not just by driver id.</b>
|
||||
/// <c>ReprovisionRequested</c> is delivered asynchronously, so one can still be in flight from an old
|
||||
/// incarnation when <c>HandleRestartDriver</c> has already stopped that child and swapped in a new
|
||||
/// one under the same <c>DriverInstanceId</c>. Routing on the id alone delivers it to the NEW child,
|
||||
/// which is in <c>Connecting</c> around a healthy driver: it triggers a second, concurrent
|
||||
/// <c>InitializeAsync</c> that races the very dispose-before-overwrite the #527 teardown exists to
|
||||
/// serialise (<c>ModbusDriver._transport</c>, <c>MqttDriver._connection</c>).
|
||||
/// <para>
|
||||
/// 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
|
||||
/// <c>InitializeCount</c> must then not move.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
[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<Commons.Interfaces.NodeDiagnosticsSnapshot>(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");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The verified negative, pinned so this plan's edits to the restart paths cannot break the one that
|
||||
/// was already correct. Force Reconnect (<c>ReconnectDriver</c>) Tells <c>ForceReconnect</c> to the
|
||||
|
||||
@@ -261,11 +261,19 @@ internal sealed class RecordingSubscribableDriver : IDriver, ISubscribable, IAla
|
||||
get { lock (_gate) return _lastSubscribedAlarmRefs; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// When set, <see cref="InitializeAsync"/> awaits it instead of completing inline — so the actor
|
||||
/// stays in <c>Connecting</c> for as long as the test wants. Needed because several defects only
|
||||
/// bite a child that has NOT yet connected (<c>InitializeRequested</c> is unhandled in
|
||||
/// <c>Connected</c>, so a test built on an already-connected child passes vacuously).
|
||||
/// </summary>
|
||||
public TaskCompletionSource? InitializeGate { get; set; }
|
||||
|
||||
/// <inheritdoc />
|
||||
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);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -359,9 +367,18 @@ internal sealed class RecordingSubscribableDriverFactory : IDriverFactory
|
||||
private readonly Lock _gate = new();
|
||||
private readonly List<RecordingSubscribableDriver> _created = new();
|
||||
|
||||
private readonly bool _gateInitialize;
|
||||
|
||||
/// <summary>Initializes a factory that materialises exactly one driver type.</summary>
|
||||
/// <param name="supportedType">The driver type this factory answers for.</param>
|
||||
public RecordingSubscribableDriverFactory(string supportedType) => _supportedType = supportedType;
|
||||
/// <param name="gateInitialize">When true, every created driver gets an un-completed
|
||||
/// <see cref="RecordingSubscribableDriver.InitializeGate"/>, so its child actor stays in
|
||||
/// <c>Connecting</c> until the test completes it.</param>
|
||||
public RecordingSubscribableDriverFactory(string supportedType, bool gateInitialize = false)
|
||||
{
|
||||
_supportedType = supportedType;
|
||||
_gateInitialize = gateInitialize;
|
||||
}
|
||||
|
||||
/// <summary>Every driver this factory created, in creation order.</summary>
|
||||
public IReadOnlyList<RecordingSubscribableDriver> 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;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user