lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs

using Akka.Actor;
using Microsoft.EntityFrameworkCore;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using ZB.MOM.WW.OtOpcUa.ControlPlane.AdminOperations;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests;

/// <summary>
///     Review 03 §2.2: a <c>Deployment</c> row left in <c>Dispatching</c> with NO <c>NodeDeploymentState</c>
///     rows — <c>HandleDispatch</c> threw before its <c>SaveChanges</c>, or the process died between the
///     AdminOps row write and the coordinator's <c>Tell</c> — used to be recovered with an EMPTY expected
///     set, at which point the deadline hit the <c>0 == 0</c> "race" branch and returned without touching
///     the row. Every later <c>StartDeployment</c> was then refused <c>AnotherDeploymentInFlight</c> forever.
/// </summary>
public sealed class ConfigPublishCoordinatorRecoveryTests : ControlPlaneActorTestBase
{
    private static readonly RevisionHash TestRevision = RevisionHash.Parse(new string('c', 64));
    private static readonly TimeSpan NoTimeout = TimeSpan.FromMinutes(5);

    /// <summary>
    ///     A lost dispatch that is still inside the apply deadline is re-driven: PreStart re-dispatches it,
    ///     the expected set is seeded from the enabled rows, and the node's ack seals it.
    /// </summary>
    [Fact]
    public void Recovered_row_with_no_expected_set_is_redispatched_and_then_seals()
    {
        var dbFactory = NewInMemoryDbFactory();
        var deploymentId = SeedDeployment(dbFactory, DeploymentStatus.Dispatching, DateTime.UtcNow);
        SeedNodes(dbFactory, "site-a-1:4053");

        // No DispatchDeployment is ever sent — PreStart alone has to recover it.
        var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout));

        AwaitAssert(() =>
        {
            using var db = dbFactory.CreateDbContext();
            db.NodeDeploymentStates.Select(s => s.NodeId).ShouldBe(new[] { "site-a-1:4053" });
            db.Deployments.Single().Status.ShouldBe(DeploymentStatus.AwaitingApplyAcks);
        }, duration: PresenceBudget);

        actor.Tell(new ApplyAck(deploymentId, NodeId.Parse("site-a-1:4053"),
            ApplyAckOutcome.Applied, null, CorrelationId.NewId()));

        AwaitAssert(() =>
        {
            using var db = dbFactory.CreateDbContext();
            db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed);
        }, duration: PresenceBudget);
    }

    /// <summary>
    ///     A lost dispatch that is already past the apply deadline is closed as <c>TimedOut</c> with a reason
    ///     naming the condition — never left in <c>Dispatching</c> — and the next deployment proceeds
    ///     (positive control: it is dispatched, acked and sealed through the same coordinator).
    /// </summary>
    [Fact]
    public void Recovered_row_past_deadline_with_no_expected_set_is_timed_out_not_wedged()
    {
        var dbFactory = NewInMemoryDbFactory();
        var staleId = SeedDeployment(dbFactory, DeploymentStatus.Dispatching, DateTime.UtcNow.AddMinutes(-10));
        SeedNodes(dbFactory, "site-a-1:4053");

        var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout));

        AwaitAssert(() =>
        {
            using var db = dbFactory.CreateDbContext();
            var row = db.Deployments.Single(d => d.DeploymentId == staleId.Value);
            row.Status.ShouldBe(DeploymentStatus.TimedOut);
            row.FailureReason.ShouldNotBeNull();
            row.FailureReason.ShouldContain("lost");
        }, duration: PresenceBudget);

        // Positive control: the coordinator is not wedged — a fresh deployment runs end to end.
        var nextId = SeedDeployment(dbFactory, DeploymentStatus.Dispatching, DateTime.UtcNow);
        actor.Tell(new DispatchDeployment(nextId, TestRevision, CorrelationId.NewId()));
        AwaitAssert(() =>
        {
            using var db = dbFactory.CreateDbContext();
            db.Deployments.Single(d => d.DeploymentId == nextId.Value).Status.ShouldBe(DeploymentStatus.AwaitingApplyAcks);
        }, duration: PresenceBudget);

        actor.Tell(new ApplyAck(nextId, NodeId.Parse("site-a-1:4053"),
            ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
        AwaitAssert(() =>
        {
            using var db = dbFactory.CreateDbContext();
            db.Deployments.Single(d => d.DeploymentId == nextId.Value).Status.ShouldBe(DeploymentStatus.Sealed);
        }, duration: PresenceBudget);
    }

    /// <summary>
    ///     An <c>AwaitingApplyAcks</c> row with no rows (the process died between the status flip and the
    ///     empty-fleet seal) is re-driven too, and with no enabled nodes it seals empty exactly as a fresh
    ///     dispatch would — the recovery keys on the missing expected set, not on the status value.
    /// </summary>
    [Fact]
    public void Recovered_AwaitingApplyAcks_row_with_no_rows_and_no_fleet_seals_empty()
    {
        var dbFactory = NewInMemoryDbFactory();
        SeedDeployment(dbFactory, DeploymentStatus.AwaitingApplyAcks, DateTime.UtcNow);
        Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout));

        AwaitAssert(() =>
        {
            using var db = dbFactory.CreateDbContext();
            db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed);
            db.NodeDeploymentStates.ShouldBeEmpty();
        }, duration: PresenceBudget);
    }

    /// <summary>
    ///     The brief's scenario end to end: <c>HandleDispatch</c> throws once (a transient fault seeding the
    ///     expected set). The coordinator restarts, recovers the <c>Dispatching</c> row, re-dispatches it and
    ///     the deployment completes; a second <c>StartDeployment</c> is then Accepted (positive control: a
    ///     normal deploy seals).
    /// </summary>
    [Fact]
    public async Task HandleDispatch_throw_is_recovered_and_the_next_StartDeployment_succeeds()
    {
        var dbName = Guid.NewGuid().ToString("N");
        var adminFactory = NewInMemoryDbFactory(dbName);
        var faulting = new OneShotFaultingDbFactory(NewInMemoryDbFactory(dbName));
        SeedNodes(adminFactory, "site-a-1:4053");

        var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(faulting, NoTimeout));
        var admin = Sys.ActorOf(AdminOperationsActor.Props(adminFactory, coordinator, Enumerable.Empty<IDriverProbe>()));

        // Arm the fault AFTER PreStart has run (it would otherwise fault the recovery read itself and
        // the actor would be stopped, not restarted). Identify is processed after creation, so a
        // resolved selection proves PreStart completed.
        await Sys.ActorSelection(coordinator.Path).ResolveOne(PresenceBudget);
        faulting.FailNext();
        admin.Tell(new StartDeployment("joe", CorrelationId.NewId()));
        var first = ExpectMsg<StartDeploymentResult>(PresenceBudget);
        first.Outcome.ShouldBe(StartDeploymentOutcome.Accepted);
        first.DeploymentId.ShouldNotBeNull();

        // The fault fired, the actor restarted, PreStart re-dispatched: the expected set lands.
        AwaitAssert(() =>
        {
            faulting.Faulted.ShouldBeTrue();
            using var db = adminFactory.CreateDbContext();
            db.NodeDeploymentStates.Count(s => s.DeploymentId == first.DeploymentId.Value.Value).ShouldBe(1);
            db.Deployments.Single().Status.ShouldBe(DeploymentStatus.AwaitingApplyAcks);
        }, duration: PresenceBudget);

        coordinator.Tell(new ApplyAck(first.DeploymentId.Value, NodeId.Parse("site-a-1:4053"),
            ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
        AwaitAssert(() =>
        {
            using var db = adminFactory.CreateDbContext();
            db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed);
        }, duration: PresenceBudget);

        // Positive control: nothing is wedged — the next deploy is Accepted and seals too.
        admin.Tell(new StartDeployment("joe", CorrelationId.NewId()));
        var second = ExpectMsg<StartDeploymentResult>(PresenceBudget);
        second.Outcome.ShouldBe(StartDeploymentOutcome.Accepted);
        coordinator.Tell(new ApplyAck(second.DeploymentId!.Value, NodeId.Parse("site-a-1:4053"),
            ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
        AwaitAssert(() =>
        {
            using var db = adminFactory.CreateDbContext();
            db.Deployments.Count(d => d.Status == DeploymentStatus.Sealed).ShouldBe(2);
        }, duration: PresenceBudget);
    }

    /// <summary>
    ///     Fix round 1 / I7 — the window PreStart cannot see: the Dispatching row is committed AFTER the
    ///     coordinator's PreStart ran, and the coordinator is never told (the admin singleton died between
    ///     its commit and the Tell). Nothing would ever close it. The in-flight check now closes a
    ///     Dispatching row with NO expected-ack rows once it is past the apply deadline, so the next
    ///     StartDeployment is Accepted (and seals — positive control on the pipeline).
    /// </summary>
    [Fact]
    public async Task Lost_dispatch_committed_after_PreStart_is_closed_by_the_next_StartDeployment()
    {
        var dbFactory = NewInMemoryDbFactory();
        SeedNodes(dbFactory, "site-a-1:4053");
        var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout));
        await Sys.ActorSelection(coordinator.Path).ResolveOne(PresenceBudget);   // PreStart saw nothing

        var stale = SeedDeployment(dbFactory, DeploymentStatus.Dispatching, DateTime.UtcNow.AddMinutes(-10));
        var admin = Sys.ActorOf(AdminOperationsActor.Props(
            dbFactory, coordinator, Enumerable.Empty<IDriverProbe>(), applyDeadline: NoTimeout));

        admin.Tell(new StartDeployment("joe", CorrelationId.NewId()));
        var result = ExpectMsg<StartDeploymentResult>(PresenceBudget);
        result.Outcome.ShouldBe(StartDeploymentOutcome.Accepted, result.Message);

        using (var db = dbFactory.CreateDbContext())
        {
            var row = db.Deployments.Single(d => d.DeploymentId == stale.Value);
            row.Status.ShouldBe(DeploymentStatus.TimedOut);
            row.FailureReason.ShouldNotBeNull();
            row.FailureReason.ShouldContain("lost");
        }

        coordinator.Tell(new ApplyAck(result.DeploymentId!.Value, NodeId.Parse("site-a-1:4053"),
            ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
        AwaitAssert(() =>
        {
            using var db = dbFactory.CreateDbContext();
            db.Deployments.Single(d => d.DeploymentId == result.DeploymentId.Value.Value).Status
                .ShouldBe(DeploymentStatus.Sealed);
        }, duration: PresenceBudget);
    }

    /// <summary>Control for the closure: the same row INSIDE the deadline is still in flight and still refuses.</summary>
    [Fact]
    public void Dispatching_row_inside_the_deadline_still_refuses_a_new_deployment()
    {
        var dbFactory = NewInMemoryDbFactory();
        var coordinator = CreateTestProbe();
        var pending = SeedDeployment(dbFactory, DeploymentStatus.Dispatching, DateTime.UtcNow.AddSeconds(-5));
        var admin = Sys.ActorOf(AdminOperationsActor.Props(
            dbFactory, coordinator.Ref, Enumerable.Empty<IDriverProbe>(), applyDeadline: NoTimeout));

        admin.Tell(new StartDeployment("joe", CorrelationId.NewId()));
        var result = ExpectMsg<StartDeploymentResult>(PresenceBudget);
        result.Outcome.ShouldBe(StartDeploymentOutcome.AnotherDeploymentInFlight);
        result.DeploymentId.ShouldBe(pending);

        using var db = dbFactory.CreateDbContext();
        db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Dispatching);
    }

    /// <summary>
    ///     Fix round 1 minor — a late or duplicated <c>DispatchDeployment</c> for a row that is already
    ///     terminal (it raced the lost-dispatch closure above, or a re-delivery after a seal) must not flip
    ///     it back to AwaitingApplyAcks or seed an expected set. Positive control: a live row dispatched right
    ///     after it IS seeded (so the first message was processed, not merely not-yet-processed).
    /// </summary>
    [Theory]
    [InlineData(DeploymentStatus.Sealed)]
    [InlineData(DeploymentStatus.TimedOut)]
    [InlineData(DeploymentStatus.PartiallyFailed)]
    public void Dispatch_of_a_terminal_row_is_ignored(DeploymentStatus terminal)
    {
        var dbFactory = NewInMemoryDbFactory();
        SeedNodes(dbFactory, "site-a-1:4053");
        var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout));

        var done = SeedDeployment(dbFactory, terminal, DateTime.UtcNow.AddMinutes(-1));
        coordinator.Tell(new DispatchDeployment(done, TestRevision, CorrelationId.NewId()));
        var live = SeedDeployment(dbFactory, DeploymentStatus.Dispatching, DateTime.UtcNow);
        coordinator.Tell(new DispatchDeployment(live, TestRevision, CorrelationId.NewId()));

        AwaitAssert(() =>
        {
            using var db = dbFactory.CreateDbContext();
            db.Deployments.Single(d => d.DeploymentId == live.Value).Status.ShouldBe(DeploymentStatus.AwaitingApplyAcks);
        }, duration: PresenceBudget);

        using var verify = dbFactory.CreateDbContext();
        verify.Deployments.Single(d => d.DeploymentId == done.Value).Status.ShouldBe(terminal);
        verify.NodeDeploymentStates.Count(s => s.DeploymentId == done.Value).ShouldBe(0);
    }

    /// <summary>
    ///     Fix round 1 minor — status transitions update a blob-free stub (key + RowVersion) instead of
    ///     materialising the multi-MB row. Proven on the REAL provider, where the stub's RowVersion is the
    ///     optimistic-concurrency predicate: dispatch → ack → Sealed lands, and the artifact bytes are
    ///     untouched by the partial UPDATEs.
    /// </summary>
    [Fact]
    [Trait("Category", "SchemaCompliance")]
    public void On_sql_server_status_transitions_do_not_touch_the_artifact_blob()
    {
        using var sql = new SqlServerTestDatabase();
        var factory = sql.Factory();
        SeedNodes(factory, "site-a-1:4053");
        var blob = Enumerable.Range(0, 4096).Select(i => (byte)i).ToArray();
        var id = DeploymentId.NewId();
        using (var db = sql.NewContext())
        {
            db.Deployments.Add(new Deployment
            {
                DeploymentId = id.Value, RevisionHash = TestRevision.Value, Status = DeploymentStatus.Dispatching,
                CreatedBy = "test", ArtifactBlob = blob,
            });
            db.SaveChanges();
        }

        var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(factory, NoTimeout));
        AwaitAssert(() =>
        {
            using var db = sql.NewContext();
            db.Deployments.AsNoTracking().Single().Status.ShouldBe(DeploymentStatus.AwaitingApplyAcks);
        }, duration: PresenceBudget);   // PreStart re-drives the un-dispatched row
        coordinator.Tell(new ApplyAck(id, NodeId.Parse("site-a-1:4053"), ApplyAckOutcome.Applied, null, CorrelationId.NewId()));

        AwaitAssert(() =>
        {
            using var db = sql.NewContext();
            var row = db.Deployments.AsNoTracking().Single();
            row.Status.ShouldBe(DeploymentStatus.Sealed);
            row.SealedAtUtc.ShouldNotBeNull();
            row.ArtifactBlob.ShouldBe(blob);
        }, duration: PresenceBudget);
    }

    /// <summary>
    ///     Re-review 1 — the reopen race. AdminOps' lost-dispatch closure commits TimedOut AFTER the coordinator's
    ///     terminal guard read the row as Dispatching but BEFORE the AwaitingApplyAcks flip. The flip is
    ///     conditioned on the guard's RowVersion, so on SQL Server it becomes a concurrency conflict and the row
    ///     stays TimedOut with no expected set. Positive control: the coordinator is not wedged — the next live
    ///     dispatch is seeded and flipped.
    /// </summary>
    [Fact]
    [Trait("Category", "SchemaCompliance")]
    public async Task On_sql_server_a_closure_between_guard_and_flip_is_not_reopened()
    {
        using var sql = new SqlServerTestDatabase();
        var inner = sql.Factory();
        SeedNodes(inner, "site-a-1:4053");
        var interposer = new InterposingDbFactory(inner);
        var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(interposer, NoTimeout));
        await Sys.ActorSelection(coordinator.Path).ResolveOne(PresenceBudget);   // PreStart ran over an empty table

        // The race proper: a new Dispatching row whose dispatch is interleaved with an AdminOps-style closure.
        var victim = SeedDeployment(inner, DeploymentStatus.Dispatching, DateTime.UtcNow);
        interposer.OnSecondCreate(() =>
        {
            using var db = sql.NewContext();
            db.Database.ExecuteSqlRaw(
                "UPDATE Deployment SET Status = {0}, FailureReason = N'closed as lost' WHERE DeploymentId = {1}",
                (int)DeploymentStatus.TimedOut, victim.Value);
        });
        coordinator.Tell(new DispatchDeployment(victim, TestRevision, CorrelationId.NewId()));
        AwaitAssert(() => interposer.Fired.ShouldBeTrue(), duration: PresenceBudget);

        // Positive control + ordering: a live row dispatched after it IS flipped, so the victim's dispatch was processed.
        var live = SeedDeployment(inner, DeploymentStatus.Dispatching, DateTime.UtcNow);
        coordinator.Tell(new DispatchDeployment(live, TestRevision, CorrelationId.NewId()));
        AwaitAssert(() =>
        {
            using var db = sql.NewContext();
            db.Deployments.AsNoTracking().Single(d => d.DeploymentId == live.Value).Status
                .ShouldBe(DeploymentStatus.AwaitingApplyAcks);
        }, duration: PresenceBudget);

        using var verify = sql.NewContext();
        verify.Deployments.AsNoTracking().Single(d => d.DeploymentId == victim.Value).Status
            .ShouldBe(DeploymentStatus.TimedOut, "the closure must not be reopened");
        verify.NodeDeploymentStates.Count(s => s.DeploymentId == victim.Value).ShouldBe(0);
    }

    /// <summary>Runs an action just before handing out the SECOND context after arming (HandleDispatch: guard, then flip).</summary>
    private sealed class InterposingDbFactory(IDbContextFactory<OtOpcUaConfigDbContext> inner)
        : IDbContextFactory<OtOpcUaConfigDbContext>
    {
        private Action? _action;
        private int _countdown;
        public volatile bool Fired;

        public void OnSecondCreate(Action action)
        {
            _action = action;
            Volatile.Write(ref _countdown, 2);
        }

        public OtOpcUaConfigDbContext CreateDbContext()
        {
            if (Volatile.Read(ref _countdown) > 0 && Interlocked.Decrement(ref _countdown) == 0)
            {
                _action?.Invoke();
                Fired = true;
            }
            return inner.CreateDbContext();
        }
    }

    private static DeploymentId SeedDeployment(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory, DeploymentStatus status, DateTime createdAtUtc)
    {
        var id = DeploymentId.NewId();
        using var db = dbFactory.CreateDbContext();
        db.Deployments.Add(new Deployment
        {
            DeploymentId = id.Value,
            RevisionHash = TestRevision.Value,
            Status = status,
            CreatedBy = "test",
            CreatedAtUtc = createdAtUtc,
        });
        db.SaveChanges();
        return id;
    }

    private static void SeedNodes(IDbContextFactory<OtOpcUaConfigDbContext> dbFactory, params string[] nodeIds)
    {
        using var db = dbFactory.CreateDbContext();
        db.ServerClusters.Add(new ServerCluster
        {
            ClusterId = "SITE-A", Name = "Site A", Enterprise = "zb", Site = "site-a",
            NodeCount = (byte)nodeIds.Length, RedundancyMode = RedundancyMode.None, CreatedBy = "test",
        });
        foreach (var nodeId in nodeIds)
        {
            db.ClusterNodes.Add(new ClusterNode
            {
                NodeId = nodeId, ClusterId = "SITE-A", Host = nodeId.Split(':')[0],
                ApplicationUri = $"urn:OtOpcUa:{nodeId}", Enabled = true, CreatedBy = "test",
            });
        }
        db.SaveChanges();
    }

    /// <summary>Throws from exactly one <c>CreateDbContext</c> call after <see cref="FailNext"/>.</summary>
    private sealed class OneShotFaultingDbFactory(IDbContextFactory<OtOpcUaConfigDbContext> inner)
        : IDbContextFactory<OtOpcUaConfigDbContext>
    {
        private int _armed;
        public bool Faulted { get; private set; }

        public void FailNext() => Interlocked.Exchange(ref _armed, 1);

        public OtOpcUaConfigDbContext CreateDbContext()
        {
            if (Interlocked.Exchange(ref _armed, 0) == 1)
            {
                Faulted = true;
                throw new InvalidOperationException("injected transient fault seeding the expected set");
            }
            return inner.CreateDbContext();
        }
    }
}