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

using System.Collections.Concurrent;
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 P2: <c>WriteAttribute</c> was a <c>ReceiveAsync</c>, which suspends the mailbox for the whole
///     device round-trip (up to the 5 s bound). Writes therefore serialised — the SECOND of two operator writes to
///     a ~3 s device missed the host's 8 s Ask and was reverted although the device took it — and every data
///     change queued behind the write. Writes now run off the mailbox under a per-driver in-flight cap; the
///     subscribe/unsubscribe pair that must serialise against the subscription handle stays <c>ReceiveAsync</c>.
/// </summary>
public sealed class DriverInstanceActorConcurrentWriteTests : RuntimeActorTestBase
{
    private static readonly TimeSpan WriteLatency = TimeSpan.FromMilliseconds(1200);

    /// <summary>
    ///     Polls for <paramref name="count"/> writes inside the driver with an EXPLICIT short interval. Akka's
    ///     <c>AwaitCondition</c> without one polls every <c>max / 10</c> of the DILATED budget — 1.5 s locally, 4.5 s
    ///     under CI's <c>AKKA_TEST_TIMEFACTOR=3</c>. The write under test is already held inside the driver by then,
    ///     and it was bounded by <c>DriverInstanceActor.RunWriteAsync</c>'s fixed (undilated) 5 s token, so one
    ///     missed first poll plus the test's own settle window cancelled it ("write timeout") before
    ///     <c>ReleaseWrites()</c>: <c>A_queued_write_older_than_the_budget…</c> failed every CI run that hit it
    ///     (hosted runs 2215 and 2260), and failed locally at timefactor 3. Since lmxopcua#774 that token follows
    ///     the actor's write timeout, and <see cref="Connect"/> sets it to the dilated presence budget, so a held
    ///     write no longer races an undilated bound at all; the short interval stays to keep the tests fast.
    /// </summary>
    private void AwaitWritesEntered(SlowWritableDriver driver, int count) =>
        AwaitCondition(() => driver.WritesEntered == count, PresenceBudget, TimeSpan.FromMilliseconds(20));

    /// <summary>
    ///     Spawns and connects the actor. Unless a test passes its own <paramref name="writeTimeout"/>, the write
    ///     bound is the DILATED presence budget: these tests hold writes inside the driver and release them
    ///     explicitly, and are not about the bound — the production 5 s default would be an undilated deadline
    ///     racing every held write under CI's timefactor (the 342a825d failure mode).
    /// </summary>
    private (IActorRef Actor, Akka.TestKit.TestProbe Parent) Connect(
        IDriver driver, int? maxWritesInFlight = null, TimeSpan? queuedWriteBudget = null, TimeSpan? writeTimeout = null,
        TimeProvider? writeClock = null)
    {
        var parent = CreateTestProbe();
        var actor = parent.ChildActorOf(DriverInstanceActor.Props(
            driver, commsLossTimeout: TimeSpan.Zero, maxWritesInFlight: maxWritesInFlight,
            queuedWriteBudget: queuedWriteBudget, writeTimeout: writeTimeout ?? Dilated(PresenceBudget),
            writeClock: writeClock));
        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
        parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(m => m.Connected, PresenceBudget);
        return (actor, parent);
    }

    [Fact]
    public async Task Two_concurrent_slow_writes_both_complete_within_one_device_round_trip()
    {
        var driver = new SlowWritableDriver { Latency = WriteLatency };
        var (actor, _) = Connect(driver);

        var sw = Stopwatch.StartNew();
        var askBudget = TimeSpan.FromSeconds(8); // the host's Ask
        var first = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
            new DriverInstanceActor.WriteAttribute("tag-1", 1), askBudget);
        var second = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
            new DriverInstanceActor.WriteAttribute("tag-2", 2), askBudget);
        var results = await Task.WhenAll(first, second);
        sw.Stop();

        results.ShouldAllBe(r => r.Success);
        driver.Writes.Select(w => w.FullReference).ShouldBe(new[] { "tag-1", "tag-2" }, ignoreOrder: true);
        // Both were inside the driver at once — that is the proof; the wall clock is only a sanity ceiling
        // (serialised writes would take two latencies, and the host's Ask is 8 s).
        driver.MaxConcurrentWrites.ShouldBe(2, "the second write waited for the first — writes are still serialised on the mailbox");
        sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(8));
    }

    [Fact]
    public async Task Data_changes_are_delivered_while_a_write_is_pending()
    {
        var driver = new SlowWritableDriver { HoldWrites = true };
        var (actor, parent) = Connect(driver);
        await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
            new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)), PresenceBudget);

        var pending = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
            new DriverInstanceActor.WriteAttribute("tag-a", 7), PresenceBudget);
        AwaitWritesEntered(driver, 1);

        driver.FireDataChange("tag-a", 3.5);
        var published = parent.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(PresenceBudget);
        published.Items.Single().Value.ShouldBe(3.5);
        pending.IsCompleted.ShouldBeFalse("the write is still held — the data change did not wait behind it");

        // Positive control: releasing the device completes the write with its real outcome.
        driver.ReleaseWrites();
        (await pending).Success.ShouldBeTrue();
    }

    [Fact]
    public async Task Writes_beyond_the_in_flight_cap_queue_and_drain_in_order()
    {
        var driver = new SlowWritableDriver { HoldWrites = true };
        var (actor, _) = Connect(driver, maxWritesInFlight: 2);

        var asks = Enumerable.Range(0, 4)
            .Select(i => actor.Ask<DriverInstanceActor.WriteAttributeResult>(
                new DriverInstanceActor.WriteAttribute($"tag-{i}", i), PresenceBudget))
            .ToArray();

        AwaitWritesEntered(driver, 2);
        // Settle: a third write entering the driver would show up here.
        await Task.Delay(300);
        driver.WritesEntered.ShouldBe(2, "the cap let a third write into the driver");

        driver.ReleaseWrites();
        var results = await Task.WhenAll(asks);
        results.ShouldAllBe(r => r.Success);
        // Entry order: the first two slots went to the first two writes, the queued pair entered only after a
        // slot freed. Within each pair the order is unspecified — distinct tags run concurrently on the pool;
        // only SAME-tag order is guaranteed (pinned by the same-tag test below).
        var entered = driver.EnteredRefs;
        entered.Count.ShouldBe(4);
        entered.Take(2).ShouldBe(new[] { "tag-0", "tag-1" }, ignoreOrder: true);
        entered.Skip(2).ShouldBe(new[] { "tag-2", "tag-3" }, ignoreOrder: true);
    }

    /// <summary>
    ///     Fix round 1: writes to the SAME tag must keep their order — two concurrent writes v1, v2 could land as
    ///     v1. A same-tag write waits behind its tag while other tags still run concurrently.
    /// </summary>
    [Fact]
    public async Task Writes_to_the_same_tag_are_serialised_in_order_while_other_tags_run_concurrently()
    {
        var driver = new SlowWritableDriver { HoldWrites = true };
        var (actor, _) = Connect(driver, maxWritesInFlight: 4);

        var first = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
            new DriverInstanceActor.WriteAttribute("tag-a", 1), PresenceBudget);
        AwaitWritesEntered(driver, 1);
        var second = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
            new DriverInstanceActor.WriteAttribute("tag-a", 2), PresenceBudget);
        // Positive control: a DIFFERENT tag is not held behind tag-a.
        var other = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
            new DriverInstanceActor.WriteAttribute("tag-b", 9), PresenceBudget);
        AwaitWritesEntered(driver, 2);
        driver.EnteredRefs.ShouldBe(new[] { "tag-a", "tag-b" });

        await Task.Delay(300);
        driver.WritesEntered.ShouldBe(2, "the second tag-a write entered the driver while the first was still inside it");
        second.IsCompleted.ShouldBeFalse();

        driver.ReleaseWrites();
        (await Task.WhenAll(first, second, other)).ShouldAllBe(r => r.Success);
        driver.Writes.Where(w => w.FullReference == "tag-a").Select(w => w.Value).ShouldBe(new object[] { 1, 2 });
    }

    /// <summary>
    ///     Fix round 1: a write that sat in the queue longer than the host's Ask budget has already been reported
    ///     as a timeout (and reverted) upstream — writing it late would push a stale value to the device.
    /// </summary>
    [Fact]
    public async Task A_queued_write_older_than_the_budget_is_failed_BadTimeout_instead_of_written_late()
    {
        var driver = new SlowWritableDriver { HoldWrites = true };
        var (actor, _) = Connect(driver, maxWritesInFlight: 1, queuedWriteBudget: TimeSpan.FromMilliseconds(300));

        var inFlight = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
            new DriverInstanceActor.WriteAttribute("tag-0", 0), PresenceBudget);
        AwaitWritesEntered(driver, 1);
        var stale = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
            new DriverInstanceActor.WriteAttribute("tag-1", 1), PresenceBudget);

        await Task.Delay(600);
        driver.ReleaseWrites();

        (await inFlight).Success.ShouldBeTrue();
        var staleResult = await stale;
        staleResult.Success.ShouldBeFalse();
        staleResult.Reason!.ShouldContain("800A0000", customMessage: "BadTimeout");
        driver.EnteredRefs.ShouldBe(new[] { "tag-0" }, "the stale write must never reach the device");
    }

    /// <summary>
    ///     Fix round 2: a token-ignoring hung write held its tag forever, so every later write to that tag piled up
    ///     in the queue. The write is bounded pessimistically; on overrun it answers BadTimeout and releases its
    ///     slot and tag, and a later write to the same tag reaches the device.
    /// </summary>
    [Fact]
    public async Task A_hung_token_ignoring_write_is_abandoned_and_releases_its_tag_for_a_later_write()
    {
        // ONLY the first write wedges (ignoring its token); every later write completes at once. The fixture used
        // to wedge EVERY write on one shared release gate, so the later write was ALSO token-ignoring and the
        // 300 ms pessimistic bound abandoned it too while the test was still polling/asserting before
        // ReleaseWrites() — its reason was "write did not complete within" (the bound), not "queued for" (the
        // queue budget), and it had entered the driver, so the tag HAD been released. Fixture defect, not code.
        var driver = new SlowWritableDriver { HangFirstWriteIgnoringCancellation = true };
        var (actor, _) = Connect(driver, maxWritesInFlight: 1, writeTimeout: TimeSpan.FromMilliseconds(300));

        var hung = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
            new DriverInstanceActor.WriteAttribute("tag-a", 1), PresenceBudget);
        AwaitWritesEntered(driver, 1);

        var hungResult = await hung;
        hungResult.Success.ShouldBeFalse();
        hungResult.Reason!.ShouldContain("800A0000", customMessage: "BadTimeout");
        hungResult.Reason!.ShouldContain("within 300 ms", customMessage: "the abandon reason names the real bound (it used to round 300 ms down to \"0s\")");
        driver.MaxConcurrentWrites.ShouldBe(1);

        // The tag (and the single slot) are free again: a later tag-a write reaches the device even though the
        // abandoned one is still inside it.
        var later = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
            new DriverInstanceActor.WriteAttribute("tag-a", 2), PresenceBudget);
        AwaitWritesEntered(driver, 2);
        driver.EnteredRefs.ShouldBe(new[] { "tag-a", "tag-a" });
        var laterResult = await later;
        laterResult.Success.ShouldBeTrue(laterResult.Reason);
        driver.MaxConcurrentWrites.ShouldBe(2, "positive control: the abandoned write really is still wedged inside the driver");
        driver.Writes.Select(w => w.Value).ShouldBe(new object[] { 2 }, "only the later write reached the device's commit");
    }

    /// <summary>
    ///     lmxopcua#774: the write's cancellation token was a hard-coded 5 s, so a configured write timeout above
    ///     5 s never took effect. On a manual clock: a write held past 6 s under a 10 s write timeout still
    ///     succeeds when the device answers. (Mutation-checked: a fixed 5 s token on the same clock fails this
    ///     with "write timeout".)
    /// </summary>
    [Fact]
    public async Task A_write_longer_than_5s_but_within_the_configured_write_timeout_succeeds()
    {
        var clock = new ManualTimeProvider();
        var driver = new SlowWritableDriver { HoldWrites = true };
        var (actor, _) = Connect(driver, writeTimeout: TimeSpan.FromSeconds(10), writeClock: clock);

        var write = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
            new DriverInstanceActor.WriteAttribute("tag-a", 1), PresenceBudget);
        AwaitWritesEntered(driver, 1);
        AwaitWriteBoundsArmed(clock);

        clock.Advance(TimeSpan.FromSeconds(6));
        await Task.Delay(100);
        write.IsCompleted.ShouldBeFalse("the write was cut off before its configured 10 s bound");
        driver.ReleaseWrites();

        var result = await write;
        result.Success.ShouldBeTrue(result.Reason);
        driver.Writes.Select(w => w.FullReference).ShouldBe(new[] { "tag-a" });
    }

    /// <summary>
    ///     Positive control for the test above: the configured write timeout DOES bound the write — at 10 s its
    ///     token is cancelled and a token-honouring driver answers "write timeout" (not the abandon path).
    /// </summary>
    [Fact]
    public async Task The_configured_write_timeout_cancels_the_write_when_it_elapses()
    {
        var clock = new ManualTimeProvider();
        var driver = new SlowWritableDriver { HoldWrites = true };
        var (actor, _) = Connect(driver, writeTimeout: TimeSpan.FromSeconds(10), writeClock: clock);

        var write = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
            new DriverInstanceActor.WriteAttribute("tag-a", 1), PresenceBudget);
        AwaitWritesEntered(driver, 1);
        AwaitWriteBoundsArmed(clock);

        clock.Advance(TimeSpan.FromSeconds(10));
        var result = await write;
        result.Success.ShouldBeFalse();
        result.Reason.ShouldBe("write timeout");
        driver.Writes.ShouldBeEmpty();
    }

    /// <summary>
    ///     A token-IGNORING write is abandoned <see cref="DriverInstanceActor.WriteAbandonGrace"/> after its token
    ///     is cancelled, not at the same instant (a token-honouring driver gets the grace to answer itself).
    /// </summary>
    [Fact]
    public async Task A_token_ignoring_write_is_abandoned_a_grace_after_the_configured_write_timeout()
    {
        var clock = new ManualTimeProvider();
        var driver = new SlowWritableDriver { HangFirstWriteIgnoringCancellation = true };
        var (actor, _) = Connect(driver, writeTimeout: TimeSpan.FromSeconds(10), writeClock: clock);

        var write = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
            new DriverInstanceActor.WriteAttribute("tag-a", 1), PresenceBudget);
        AwaitWritesEntered(driver, 1);
        AwaitWriteBoundsArmed(clock);

        clock.Advance(TimeSpan.FromSeconds(10));
        await Task.Delay(100);
        write.IsCompleted.ShouldBeFalse("abandoned at the cancel instant — no grace for the driver to answer");

        clock.Advance(DriverInstanceActor.WriteAbandonGrace(TimeSpan.FromSeconds(10)));
        var result = await write;
        result.Success.ShouldBeFalse();
        result.Reason!.ShouldContain("800A0000", customMessage: "BadTimeout");
        result.Reason!.ShouldContain("within 10000 ms");
    }

    /// <summary>Waits until the write task has armed BOTH of its bounds on <paramref name="clock"/> (the token
    /// cancel and the abandon), so an Advance cannot run ahead of them. Counts the timers rather than matching their
    /// due times, so a bound armed at the WRONG time fails the test's own assertion instead of this wait.</summary>
    private void AwaitWriteBoundsArmed(ManualTimeProvider clock) =>
        AwaitCondition(() => clock.ArmedTimers == 2, PresenceBudget, TimeSpan.FromMilliseconds(20));

    [Fact]
    public async Task Queued_writes_fail_fast_when_the_driver_leaves_Connected()
    {
        var driver = new SlowWritableDriver { HoldWrites = true };
        var (actor, parent) = Connect(driver, maxWritesInFlight: 1);

        var inFlight = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
            new DriverInstanceActor.WriteAttribute("tag-0", 0), PresenceBudget);
        AwaitWritesEntered(driver, 1);
        var queued = actor.Ask<DriverInstanceActor.WriteAttributeResult>(
            new DriverInstanceActor.WriteAttribute("tag-1", 1), PresenceBudget);

        actor.Tell(new DriverInstanceActor.ForceReconnect());
        parent.ExpectMsg<DriverInstanceActor.ConnectivityChanged>(m => !m.Connected, PresenceBudget);

        var queuedResult = await queued;
        queuedResult.Success.ShouldBeFalse();
        queuedResult.Reason.ShouldBe("driver not connected");
        driver.WritesEntered.ShouldBe(1, "a queued write must never reach a driver that is reconnecting");

        // The in-flight one still answers with the device's real outcome once it is released.
        driver.ReleaseWrites();
        (await inFlight).Success.ShouldBeTrue();
    }

    /// <summary>A writable + subscribable driver whose writes either take <see cref="Latency"/> or hold until released.</summary>
    private sealed class SlowWritableDriver : IDriver, IWritable, ISubscribable
    {
        private readonly StubHandle _handle = new();
        private readonly ConcurrentQueue<WriteRequest> _writes = new();
        private readonly ConcurrentQueue<string> _enteredRefs = new();
        private TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously);
        private int _entered;
        private int _inside;
        private int _maxInside;

        public TimeSpan Latency { get; init; }
        public bool HoldWrites { get; init; }
        /// <summary>When true the FIRST write never completes and ignores its cancellation token — a wedged
        /// driver; every later write takes the normal path.</summary>
        public bool HangFirstWriteIgnoringCancellation { get; init; }
        private readonly TaskCompletionSource _never = new(TaskCreationOptions.RunContinuationsAsynchronously);

        public IReadOnlyList<WriteRequest> Writes => _writes.ToArray();
        public IReadOnlyList<string> EnteredRefs => _enteredRefs.ToArray();
        public int WritesEntered => Volatile.Read(ref _entered);
        public int MaxConcurrentWrites => Volatile.Read(ref _maxInside);

        public string DriverInstanceId => "slow-writer-1";
        public string DriverType => "SlowWriter";

        public void ReleaseWrites() => _release.TrySetResult();

        public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
        public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask;
        public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask;
        public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null);

        public async Task<IReadOnlyList<WriteResult>> WriteAsync(IReadOnlyList<WriteRequest> writes, CancellationToken cancellationToken)
        {
            var seq = Interlocked.Increment(ref _entered);
            foreach (var w in writes) _enteredRefs.Enqueue(w.FullReference);
            var inside = Interlocked.Increment(ref _inside);
            int seen;
            while (inside > (seen = Volatile.Read(ref _maxInside)))
                Interlocked.CompareExchange(ref _maxInside, inside, seen);
            try
            {
                if (HangFirstWriteIgnoringCancellation && seq == 1) await _never.Task;
                else if (HoldWrites) await _release.Task.WaitAsync(cancellationToken);
                else if (Latency > TimeSpan.Zero) await Task.Delay(Latency, cancellationToken);
                foreach (var w in writes) _writes.Enqueue(w);
                return writes.Select(_ => new WriteResult(OpcUaStatus.Good)).ToList();
            }
            finally
            {
                Interlocked.Decrement(ref _inside);
            }
        }

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