fix(runtime): dispose drivers on final stop, bound PostStop shutdown (A1)
DriverInstanceActor.PostStop now bounds IDriver.ShutdownAsync by a shutdownTimeout (default 5 s, logged on overrun) and then disposes the driver (IAsyncDisposable preferred, else IDisposable) — but only on a real stop. PreRestart flags the restart so the PostStop it triggers skips the dispose, because a supervision restart re-uses the same driver object (Props closure) and the host re-initialises it. DiscoveryDriverBrowser disposes its throwaway capture/probe driver after the bounded shutdown, including after a failed or timed-out one. Review refs: 02/L1, 02/P4, 06/G-1 (host context), 07/L2. Refs #731 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
+173
@@ -0,0 +1,173 @@
|
||||
using System.Diagnostics;
|
||||
using Akka.Actor;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Types;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
|
||||
|
||||
/// <summary>
|
||||
/// Review 02 L1 / P4: a driver child that stops for good must <b>dispose</b> its <see cref="IDriver"/>
|
||||
/// (the runtime only ever called <c>ShutdownAsync</c>, so a Calculation driver's compiled-script ALCs and an
|
||||
/// OpcUaClient driver's LoggerFactory + gate + CTS leaked on every respawn), its <c>ShutdownAsync</c> must be
|
||||
/// bounded (it used to block a dispatcher thread for as long as a dead socket took to answer), and a
|
||||
/// <b>supervision restart</b> must NOT dispose — the Props closure re-uses the same driver object for the
|
||||
/// restarted incarnation, which then has to serve through it.
|
||||
/// </summary>
|
||||
public sealed class DriverInstanceActorDisposeTests : RuntimeActorTestBase
|
||||
{
|
||||
[Fact]
|
||||
public void Stopping_the_child_shuts_the_driver_down_and_then_disposes_it_exactly_once()
|
||||
{
|
||||
var driver = new DisposableRecordingDriver();
|
||||
var parent = CreateTestProbe();
|
||||
parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
|
||||
var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver));
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
AwaitCondition(() => driver.InitializeCount == 1, PresenceBudget);
|
||||
|
||||
Watch(actor);
|
||||
Sys.Stop(actor);
|
||||
ExpectTerminated(actor, PresenceBudget);
|
||||
|
||||
driver.ShutdownCount.ShouldBe(1);
|
||||
driver.DisposeCount.ShouldBe(1, "a child stopped for good must dispose the driver it wrapped");
|
||||
driver.Trace.ToArray().ShouldBe(new[] { "init", "shutdown", "dispose" }, "shutdown must land before dispose");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_supervision_restart_keeps_the_driver_undisposed_and_the_restarted_child_still_serves()
|
||||
{
|
||||
var driver = new DisposableRecordingDriver();
|
||||
var parent = CreateTestProbe();
|
||||
// The probe parent's own strategy STOPS a failing child; the production host (no strategy anywhere under
|
||||
// src/Server) gets Akka's default decider, which RESTARTS it — the route under test.
|
||||
var actor = parent.ChildActorOf(
|
||||
DriverInstanceActor.Props(driver), new OneForOneStrategy(_ => Directive.Restart));
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
// Barrier: Subscribe is only handled once Connected (it is dropped in Connecting).
|
||||
parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(m => m.Connected, PresenceBudget);
|
||||
await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
|
||||
new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)), PresenceBudget);
|
||||
|
||||
// The crash seam (see DriverHostActorRestartTests): a null snapshot NREs on the actor thread, and the
|
||||
// probe parent's default decider restarts the child in place around the SAME driver object.
|
||||
driver.FireMalformedDataChange("tag-a");
|
||||
parent.FishForMessage(m => m is DriverInstanceActor.ReprovisionRequested, PresenceBudget);
|
||||
|
||||
driver.ShutdownCount.ShouldBe(1, "PreRestart runs PostStop, which shuts the driver down");
|
||||
driver.DisposeCount.ShouldBe(0, "a restart re-uses the driver object — disposing it would kill the new incarnation");
|
||||
|
||||
// Positive control: the restarted incarnation serves through the undisposed driver. The host would
|
||||
// re-provision it exactly like this (InitializeRequested + the desired set).
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
parent.FishForMessage(m => m is DriverInstanceActor.ConnectivityChanged { Connected: true }, PresenceBudget);
|
||||
await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
|
||||
new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)), PresenceBudget);
|
||||
driver.FireDataChange("tag-a", 42.0);
|
||||
var published = (DriverInstanceActor.AttributeValueBatchPublished)parent.FishForMessage(
|
||||
m => m is DriverInstanceActor.AttributeValueBatchPublished, PresenceBudget);
|
||||
published.Items.Single().FullReference.ShouldBe("tag-a");
|
||||
published.Items.Single().Value.ShouldBe(42.0);
|
||||
|
||||
// ...and the eventual real stop still disposes it — once.
|
||||
Watch(actor);
|
||||
Sys.Stop(actor);
|
||||
ExpectTerminated(actor, PresenceBudget);
|
||||
driver.DisposeCount.ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_hung_ShutdownAsync_is_abandoned_at_the_bound_and_the_driver_is_still_disposed()
|
||||
{
|
||||
var never = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
// Ignores its token on purpose: a wedged socket/COM proxy does not honour cancellation either.
|
||||
var driver = new DisposableRecordingDriver { ShutdownBehavior = _ => never.Task };
|
||||
var bound = TimeSpan.FromMilliseconds(300);
|
||||
var parent = CreateTestProbe();
|
||||
parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
|
||||
var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver, shutdownTimeout: bound));
|
||||
actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
|
||||
AwaitCondition(() => driver.InitializeCount == 1, PresenceBudget);
|
||||
|
||||
Watch(actor);
|
||||
var sw = Stopwatch.StartNew();
|
||||
Sys.Stop(actor);
|
||||
ExpectTerminated(actor, PresenceBudget);
|
||||
sw.Stop();
|
||||
|
||||
driver.ShutdownCount.ShouldBe(1, "the shutdown was attempted (positive control for the bound)");
|
||||
never.Task.IsCompleted.ShouldBeFalse("the shutdown is still hung — the stop did not wait for it");
|
||||
sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(5), "the stop must not be pinned by the hung shutdown");
|
||||
driver.DisposeCount.ShouldBe(1, "a shutdown that timed out must not skip the dispose");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A subscribable driver that records its lifecycle and is <see cref="IAsyncDisposable"/>. Deliberately
|
||||
/// NOT also <see cref="IDisposable"/>: the actor prefers the async dispose when both exist, and a double
|
||||
/// implementation would hide a double-dispose.
|
||||
/// </summary>
|
||||
private sealed class DisposableRecordingDriver : IDriver, ISubscribable, IAsyncDisposable
|
||||
{
|
||||
private readonly StubHandle _handle = new();
|
||||
private int _initializeCount;
|
||||
private int _shutdownCount;
|
||||
private int _disposeCount;
|
||||
|
||||
public Func<CancellationToken, Task>? ShutdownBehavior { get; init; }
|
||||
|
||||
public List<string> Trace { get; } = new();
|
||||
|
||||
public int InitializeCount => Volatile.Read(ref _initializeCount);
|
||||
public int ShutdownCount => Volatile.Read(ref _shutdownCount);
|
||||
public int DisposeCount => Volatile.Read(ref _disposeCount);
|
||||
|
||||
public string DriverInstanceId => "disposable-1";
|
||||
public string DriverType => "Disposable";
|
||||
|
||||
public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
|
||||
{
|
||||
Interlocked.Increment(ref _initializeCount);
|
||||
lock (Trace) Trace.Add("init");
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
public Task ShutdownAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
Interlocked.Increment(ref _shutdownCount);
|
||||
lock (Trace) Trace.Add("shutdown");
|
||||
return ShutdownBehavior is { } behavior ? behavior(cancellationToken) : Task.CompletedTask;
|
||||
}
|
||||
|
||||
public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null);
|
||||
|
||||
public ValueTask DisposeAsync()
|
||||
{
|
||||
Interlocked.Increment(ref _disposeCount);
|
||||
lock (Trace) Trace.Add("dispose");
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
public event EventHandler<DataChangeEventArgs>? OnDataChange;
|
||||
|
||||
public Task<ISubscriptionHandle> SubscribeAsync(
|
||||
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) =>
|
||||
Task.FromResult<ISubscriptionHandle>(_handle);
|
||||
|
||||
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
public void FireDataChange(string fullRef, object? value) =>
|
||||
OnDataChange?.Invoke(this, new DataChangeEventArgs(
|
||||
_handle, fullRef, new DataValueSnapshot(value, OpcUaStatus.Good, DateTime.UtcNow, DateTime.UtcNow)));
|
||||
|
||||
/// <summary>The supervision crash seam — see <see cref="RecordingSubscribableDriver.FireMalformedDataChange"/>.</summary>
|
||||
public void FireMalformedDataChange(string fullRef) =>
|
||||
OnDataChange?.Invoke(this, new DataChangeEventArgs(_handle, fullRef, null!));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user