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

using System.Text.Json;
using Akka.Actor;
using Akka.Event;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;

namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;

/// <summary>
///     Surgical-deploys Task 7 Part B — a dispatch that arrives while an apply is in flight is
///     STASHED, not re-forwarded to <c>Self</c>.
/// </summary>
/// <remarks>
///     <para>
///         <c>Self.Forward</c> put the message back at the tail of the actor's OWN mailbox, so with
///         nothing else to do the host spun dequeue → log → enqueue for the entire fetch window —
///         thousands of iterations and thousands of log lines for one deferred deployment, burning a
///         core and starving nothing but itself of useful work. The stash holds it outside the mailbox
///         until the apply finishes.
///     </para>
///     <para>
///         Runs at <c>DEBUG</c> because the assertion IS a log line: a spin is only distinguishable
///         from a single deferral by how many times it is announced. A log event below
///         <c>akka.loglevel</c> never reaches the EventStream at all, so at the default WARNING this
///         test would count zero either way and pass vacuously.
///     </para>
/// </remarks>
public sealed class DriverHostActorApplyingStashTests()
    : RuntimeActorTestBase(AkkaTestHoconAtLevel("DEBUG"))
{
    private static readonly NodeId TestNode = NodeId.Parse("driver-test");
    private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64));
    private static readonly RevisionHash RevB = RevisionHash.Parse(new string('b', 64));

    /// <summary>Minimal parseable artifact — no drivers.</summary>
    private static byte[] ArtifactBytes() =>
        JsonSerializer.SerializeToUtf8Bytes(new { DriverInstances = Array.Empty<object>() });

    [Fact]
    public void ADispatchArrivingMidApply_IsDeferredOnce_AndAppliedAfterTheApplyCompletes()
    {
        var db = NewInMemoryDbFactory();
        var fetcher = new GatedFetcher(ArtifactBytes());
        var deploymentA = DeploymentId.NewId();
        var deploymentB = DeploymentId.NewId();

        var coordinator = CreateTestProbe();
        var actor = Sys.ActorOf(DriverHostActor.Props(
            db, TestNode, coordinator.Ref,
            localRoles: new HashSet<string> { "driver" },
            deploymentArtifactCache: new NullArtifactCache(),
            fetchAndCacheMode: true,
            artifactFetcher: fetcher));

        try
        {
            // A is in flight: the fetch has started and its task will not complete until released.
            actor.Tell(new DispatchDeployment(deploymentA, RevA, CorrelationId.NewId()));
            AwaitAssert(() => fetcher.Calls.ShouldBe(1), duration: PresenceBudget);

            var logs = CreateTestProbe();
            Sys.EventStream.Subscribe(logs.Ref, typeof(LogEvent));

            actor.Tell(new DispatchDeployment(deploymentB, RevB, CorrelationId.NewId()));

            // Collect for long enough that a forward-spin would announce itself thousands of times.
            var deferrals = 0;
            var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(1.5);
            while (DateTime.UtcNow < deadline)
            {
                if (logs.ReceiveOne(TimeSpan.FromMilliseconds(150)) is LogEvent e &&
                    e.Message?.ToString()?.Contains("deferring", StringComparison.Ordinal) == true)
                {
                    deferrals++;
                }
            }

            deferrals.ShouldBe(1);

            // Nothing was lost: B re-enters exactly once, after A's apply transitions back to Steady.
            fetcher.Release();
            coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(10)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
            coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(10)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
            fetcher.Calls.ShouldBe(2);
        }
        finally
        {
            // Never leave the fetch parked: a blocked apply would hang ActorSystem shutdown.
            fetcher.Release();
        }
    }

    /// <summary>Gates the FIRST fetch on the test; later fetches complete immediately.</summary>
    private sealed class GatedFetcher(byte[] bytes) : IDeploymentArtifactFetcher
    {
        private readonly TaskCompletionSource _gate =
            new(TaskCreationOptions.RunContinuationsAsynchronously);
        private int _calls;

        public int Calls => Volatile.Read(ref _calls);

        public void Release() => _gate.TrySetResult();

        public async Task<byte[]?> FetchAsync(string deploymentId, string expectedRevisionHash, CancellationToken ct)
        {
            if (Interlocked.Increment(ref _calls) == 1)
                await _gate.Task.ConfigureAwait(false);
            return bytes;
        }
    }

    /// <summary>Accepts every store; this class asserts on dispatch handling, not on caching.</summary>
    private sealed class NullArtifactCache : IDeploymentArtifactCache
    {
        public Task StoreAsync(string clusterId, string deploymentId, string revisionHash,
                               byte[] artifact, CancellationToken ct = default)
            => Task.CompletedTask;

        public Task<CachedDeploymentArtifact?> GetCurrentAsync(string clusterId, CancellationToken ct = default)
            => Task.FromResult<CachedDeploymentArtifact?>(null);

        public Task<CachedDeploymentArtifact?> GetCurrentUnkeyedAsync(CancellationToken ct = default)
            => Task.FromResult<CachedDeploymentArtifact?>(null);
    }
}