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

using Akka.Actor;
using Shouldly;
using Xunit;
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>
///     Gitea #488 — the periodic desired-vs-actual subscription reconcile that rides
///     <see cref="DriverInstanceActor"/>'s existing <c>health-poll</c> tick.
/// </summary>
/// <remarks>
///     <para>
///         The hole it closes: the desired ref set is re-applied on a <b>Connected transition</b> and on a
///         deploy's <c>SetDesiredSubscriptions</c>, and nowhere else. A subscription lost while the driver
///         <i>stayed</i> Connected — the shape reproduced here, a subscribe that failed and was never
///         retried — left the actor permanently subscribed to nothing while reporting Healthy.
///     </para>
///     <para>
///         All three tests drive a short <c>healthPollInterval</c> rather than the production 30 s. The
///         two absence assertions are bounded by a positive control (a counted number of ticks provably
///         processed), never by a bare sleep.
///     </para>
/// </remarks>
public sealed class DriverInstanceActorSubscriptionReconcileTests : RuntimeActorTestBase
{
    private static readonly TimeSpan FastPoll = TimeSpan.FromMilliseconds(100);

    /// <summary>
    ///     The reconcile itself: the first subscribe fails, leaving the actor Connected with a desired set
    ///     and no handle. Nothing else would ever retry it — no reconnect, no deploy — so a second
    ///     subscribe can only come from the periodic reconcile.
    /// </summary>
    [Fact]
    public void A_lost_subscription_is_re_established_while_still_connected()
    {
        var driver = new SubscribableStubDriver { SubscribeFailuresBeforeSuccess = 1 };
        var actor = Sys.ActorOf(DriverInstanceActor.Props(driver, healthPollInterval: FastPoll));

        actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions(
            new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)));
        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));

        // The connect-path subscribe runs and throws, so the handle stays null.
        AwaitCondition(() => driver.SubscribeCount >= 1, PresenceBudget);

        // Only the reconcile can produce this second attempt — and it must succeed this time.
        AwaitCondition(() => driver.SubscribeCount >= 2, PresenceBudget);
        driver.LastSubscribedRefs.ShouldBe(new[] { "tag-a" });
    }

    /// <summary>
    ///     A healthy subscription is left alone — the reconcile must not re-subscribe on every tick, which
    ///     would churn the backend every 30 s in production for no reason.
    /// </summary>
    [Fact]
    public void A_healthy_subscription_is_not_re_subscribed_on_every_tick()
    {
        var publisher = new CountingHealthPublisher();
        var driver = new SubscribableStubDriver();
        var actor = Sys.ActorOf(DriverInstanceActor.Props(
            driver, healthPublisher: publisher, healthPollInterval: FastPoll));

        actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions(
            new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)));
        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
        AwaitCondition(() => driver.SubscribeCount >= 1, PresenceBudget);

        // POSITIVE CONTROL for the absence below. Every health-poll tick publishes a snapshot, so the
        // publisher count is a direct observation of ticks PROCESSED. Waiting for several of them is what
        // makes "SubscribeCount is still 1" mean "the reconcile ran and declined" rather than "the timer
        // had not fired yet" — the latter would pass instantly and prove nothing.
        var before = publisher.Count;
        AwaitCondition(() => publisher.Count >= before + 4, PresenceBudget);

        driver.SubscribeCount.ShouldBe(1,
            "the handle is live, so the reconcile must leave it alone");
    }

    /// <summary>
    ///     A driver that is not <see cref="ISubscribable"/> can still be handed a desired set. Reconciling
    ///     it would tell <c>Subscribe</c> every tick and fail every tick, forever — so the reconcile is
    ///     gated on the capability.
    /// </summary>
    [Fact]
    public void A_non_subscribable_driver_is_never_reconciled()
    {
        var publisher = new CountingHealthPublisher();
        var driver = new StubDriver();   // IDriver only — no ISubscribable
        var actor = Sys.ActorOf(DriverInstanceActor.Props(
            driver, healthPublisher: publisher, healthPollInterval: FastPoll));

        actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions(
            new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)));
        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));

        // Same positive control: assert the reconcile stays silent across several PROCESSED ticks.
        // EventFilter bounds the absence to this block, and the reconcile announces itself at Info —
        // the first test above is the proof that this message does appear when the reconcile fires.
        EventFilter.Info(contains: "periodic reconcile").Expect(0, () =>
        {
            var before = publisher.Count;
            AwaitCondition(() => publisher.Count >= before + 4, PresenceBudget);
        });
    }

    /// <summary>Counts health-snapshot publishes, which is one per <c>health-poll</c> tick once the actor
    /// is running — the observable that turns "time passed" into "ticks were processed".</summary>
    private sealed class CountingHealthPublisher : IDriverHealthPublisher
    {
        private int _count;

        /// <summary>Number of snapshots published so far.</summary>
        public int Count => Volatile.Read(ref _count);

        /// <inheritdoc />
        public void Publish(
            string clusterId,
            string driverInstanceId,
            DriverHealth health,
            int errorCount5Min,
            DateTime? rediscoveryNeededUtc = null,
            string? rediscoveryReason = null,
            IReadOnlyList<HostConnectivityStatus>? hostStatuses = null,
            SkippedTagReport? skippedTags = null)
            => Interlocked.Increment(ref _count);
    }
}