lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorDisposeTests.cs

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!));
    }
}