using System.Text.Json; using Akka.Actor; using Microsoft.EntityFrameworkCore; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Commons.Interfaces; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet; 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.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; /// /// Issue #485 (driver-side half) — an artifact the host could not obtain must not be mistaken for a /// configuration that contains nothing. /// /// /// /// ReconcileDrivers and PushDesiredSubscriptions both read the artifact as /// …FirstOrDefault() ?? Array.Empty<byte>(). The throw path in each already /// returns early, but a row that is absent (or carries a zero-length blob) yields empty bytes that /// flow onwards as a real answer: zero driver specs, so DriverSpawnPlanner plans every /// running child for StopChild, and then an empty desired-subscription set drops each /// surviving driver's live handle. A node loses its entire field I/O and still ACKs Applied. /// /// /// The same reasoning as the address-space half applies: nothing legitimate produces a zero-length /// blob — an operator who really deletes every driver still deploys a JSON document with empty /// arrays — so "no bytes" can only mean the read did not answer. Seeding a row whose /// ArtifactBlob is empty reproduces exactly the bytes the missing-row case delivers, without /// needing to race a delete against the two reads. /// /// public sealed class DriverHostActorUnreadableArtifactTests : RuntimeActorTestBase { 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)); private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5); /// How long to let post-ACK subscription traffic settle before asserting it did NOT happen. private static readonly TimeSpan SettleWindow = TimeSpan.FromSeconds(2); private const string SpeedRef = "Plant/Modbus/dev1/speed"; /// A dispatch whose artifact reads back as no bytes must leave the running drivers — and their /// live subscriptions — exactly as they were. [Fact] public void Dispatch_whose_artifact_reads_back_empty_keeps_the_drivers_and_their_subscriptions() { var db = NewInMemoryDbFactory(); var factory = new SubscribingDriverFactory("Modbus"); var (actor, coordinator) = SpawnHostAndApply(db, SeedV3Deployment(db, RevA), factory); // Baseline: the child is up and subscribed to its one raw tag. AwaitAssert(() => factory.LastSubscribedRefs.ShouldNotBeNull()!.ShouldContain(SpeedRef), duration: Timeout); AskDiagnostics(actor).Drivers.Select(d => d.Name).ShouldContain("drv-1"); // The next dispatch's artifact comes back as no bytes at all. actor.Tell(new DispatchDeployment(SeedUnreadableDeployment(db, RevB), RevB, CorrelationId.NewId())); coordinator.ExpectMsg(Timeout); // The ACK precedes the SubscribeBulk pass, and the child's unsubscribe is a further async self-tell, // so settle before asserting an ABSENCE. SettleWindow is calibrated by // , which observes the very same // unsubscribe ARRIVING well inside it — without that control this assertion would pass on a race. AskDiagnostics(actor); // ordering barrier through the host's mailbox Thread.Sleep(SettleWindow); AskDiagnostics(actor).Drivers.Select(d => d.Name).ShouldContain("drv-1"); // not stopped factory.UnsubscribeCount.ShouldBe(0); // handle not dropped factory.LastSubscribedRefs.ShouldNotBeNull()!.ShouldContain(SpeedRef); } /// Positive control for the subscription half: a READABLE artifact that keeps the driver but /// drops its last tag genuinely does clear the live subscription — the child receives an empty desired /// set and unsubscribes. This proves both that the unsubscribe is observable through this factory and /// that it lands well within , so the absence asserted above is real. [Fact] public void Dropping_a_drivers_last_tag_does_clear_its_subscription() { var db = NewInMemoryDbFactory(); var factory = new SubscribingDriverFactory("Modbus"); var (actor, coordinator) = SpawnHostAndApply(db, SeedV3Deployment(db, RevA), factory); AwaitAssert(() => factory.LastSubscribedRefs.ShouldNotBeNull()!.ShouldContain(SpeedRef), duration: Timeout); factory.UnsubscribeCount.ShouldBe(0); actor.Tell(new DispatchDeployment(SeedTaglessDriverDeployment(db, RevB), RevB, CorrelationId.NewId())); coordinator.ExpectMsg(Timeout); AwaitAssert(() => factory.UnsubscribeCount.ShouldBe(1), duration: SettleWindow); AskDiagnostics(actor).Drivers.Select(d => d.Name).ShouldContain("drv-1"); // the driver itself stayed } /// Positive control: the guard keys on "the read gave us nothing", not on "the new config has /// fewer drivers". A READABLE artifact that genuinely drops the driver still stops it — otherwise the /// test above would pass just as happily against a host that had stopped reconciling altogether. [Fact] public void Dispatch_whose_readable_artifact_genuinely_drops_the_driver_still_stops_it() { var db = NewInMemoryDbFactory(); var factory = new SubscribingDriverFactory("Modbus"); var (actor, coordinator) = SpawnHostAndApply(db, SeedV3Deployment(db, RevA), factory); AwaitAssert(() => factory.LastSubscribedRefs.ShouldNotBeNull()!.ShouldContain(SpeedRef), duration: Timeout); // A real artifact that simply carries no drivers — an operator deleting the last driver. actor.Tell(new DispatchDeployment(SeedDriverlessDeployment(db, RevB), RevB, CorrelationId.NewId())); coordinator.ExpectMsg(Timeout); AwaitAssert( () => AskDiagnostics(actor).Drivers.Select(d => d.Name).ShouldNotContain("drv-1"), duration: Timeout); } /// /// Issue #486 — an apply that read no artifact applied nothing, so it must NOT be reported as /// Applied. Reporting success also advanced _currentRevision, and because /// HandleDispatchFromSteady short-circuits on a revision match, the node could never be /// healed by re-dispatching that revision. /// [Fact] public void Dispatch_whose_artifact_reads_back_empty_acks_Failed_and_leaves_the_revision_alone() { var db = NewInMemoryDbFactory(); var factory = new SubscribingDriverFactory("Modbus"); var (actor, coordinator) = SpawnHostAndApply(db, SeedV3Deployment(db, RevA), factory); AskDiagnostics(actor).CurrentRevision.ShouldBe(RevA); actor.Tell(new DispatchDeployment(SeedUnreadableDeployment(db, RevB), RevB, CorrelationId.NewId())); var ack = coordinator.ExpectMsg(Timeout); ack.Outcome.ShouldBe(ApplyAckOutcome.Failed); ack.FailureReason.ShouldNotBeNullOrWhiteSpace(); var snapshot = AskDiagnostics(actor); snapshot.CurrentRevision.ShouldBe(RevA); // never applied ⇒ never claimed snapshot.Drivers.Select(d => d.Name).ShouldContain("drv-1"); // …and #485 still holds } /// Issue #486 — the point of failing the apply: the SAME revision can be dispatched again and /// now actually applies, instead of being waved through by the "already at this rev" short-circuit. The /// recovered artifact adds a second driver, so a short-circuited ACK (which has no side effects) cannot /// satisfy this test — drv-2 only appears if the artifact was genuinely re-read and reconciled. [Fact] public void A_failed_apply_is_retried_not_short_circuited_once_the_artifact_is_readable_again() { var db = NewInMemoryDbFactory(); var factory = new SubscribingDriverFactory("Modbus"); var (actor, coordinator) = SpawnHostAndApply(db, SeedV3Deployment(db, RevA), factory); var deploymentId = SeedUnreadableDeployment(db, RevB); actor.Tell(new DispatchDeployment(deploymentId, RevB, CorrelationId.NewId())); coordinator.ExpectMsg(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Failed); // The ConfigDb recovers: the row now carries a real artifact, under the SAME revision. SetArtifact(db, deploymentId, TwoDriverArtifact()); actor.Tell(new DispatchDeployment(deploymentId, RevB, CorrelationId.NewId())); coordinator.ExpectMsg(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied); AwaitAssert( () => AskDiagnostics(actor).Drivers.Select(d => d.Name).ShouldContain("drv-2"), duration: Timeout); AskDiagnostics(actor).CurrentRevision.ShouldBe(RevB); } /// Spawns the host with the subscribing factory, dispatches and /// waits for the Applied ACK so the child + the initial SubscribeBulk pass are in place. private (IActorRef Actor, Akka.TestKit.TestProbe Coordinator) SpawnHostAndApply( IDbContextFactory db, DeploymentId deploymentId, IDriverFactory factory) { var coordinator = CreateTestProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( db, TestNode, coordinator.Ref, driverFactory: factory, localRoles: new HashSet { "driver" })); actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); coordinator.ExpectMsg(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied); return (actor, coordinator); } private NodeDiagnosticsSnapshot AskDiagnostics(IActorRef actor) { var probe = CreateTestProbe(); actor.Tell(new GetDiagnostics(CorrelationId.NewId()), probe.Ref); return probe.ExpectMsg(Timeout); } /// Seeds a Sealed v3 deployment: RawFolder "Plant" → DriverInstance "drv-1" (Modbus, ENABLED so a /// real child spawns) → Device "dev1" → Tag "speed" (RawPath Plant/Modbus/dev1/speed). private static DeploymentId SeedV3Deployment(IDbContextFactory db, RevisionHash rev) => SeedDeployment(db, rev, JsonSerializer.SerializeToUtf8Bytes(new { RawFolders = new[] { new { RawFolderId = "rf-plant", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } }, DriverInstances = new[] { new { DriverInstanceId = "drv-1", RawFolderId = "rf-plant", Name = "Modbus", DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = true }, }, Devices = new[] { new { DeviceId = "dev-1", DriverInstanceId = "drv-1", Name = "dev1", DeviceConfig = "{}" } }, TagGroups = Array.Empty(), Tags = new[] { new { TagId = "t-speed", DeviceId = "dev-1", TagGroupId = (string?)null, Name = "speed", DataType = "Double", AccessLevel = 1, TagConfig = "{}" }, }, })); /// A readable artifact carrying drv-1 AND drv-2, so an apply that genuinely happens is /// distinguishable from a short-circuited ACK (which spawns nothing). private static byte[] TwoDriverArtifact() => JsonSerializer.SerializeToUtf8Bytes(new { RawFolders = new[] { new { RawFolderId = "rf-plant", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } }, DriverInstances = new[] { new { DriverInstanceId = "drv-1", RawFolderId = "rf-plant", Name = "Modbus", DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = true }, new { DriverInstanceId = "drv-2", RawFolderId = "rf-plant", Name = "Modbus2", DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = true }, }, Devices = new[] { new { DeviceId = "dev-1", DriverInstanceId = "drv-1", Name = "dev1", DeviceConfig = "{}" } }, TagGroups = Array.Empty(), Tags = new[] { new { TagId = "t-speed", DeviceId = "dev-1", TagGroupId = (string?)null, Name = "speed", DataType = "Double", AccessLevel = 1, TagConfig = "{}" }, }, }); /// Replaces a seeded deployment's artifact in place — the ConfigDb becoming readable again /// without the revision changing. private static void SetArtifact( IDbContextFactory db, DeploymentId deploymentId, byte[] artifact) { // ArtifactBlob is init-only, so the row is replaced rather than mutated: delete and re-add under // the SAME id + revision. Separate SaveChanges calls so the in-memory provider does not see a // delete and an insert of the same key in one change set. using var ctx = db.CreateDbContext(); var row = ctx.Deployments.Single(d => d.DeploymentId == deploymentId.Value); var rev = row.RevisionHash; ctx.Deployments.Remove(row); ctx.SaveChanges(); ctx.Deployments.Add(new Deployment { DeploymentId = deploymentId.Value, RevisionHash = rev, Status = DeploymentStatus.Sealed, CreatedBy = "test", SealedAtUtc = DateTime.UtcNow, ArtifactBlob = artifact, }); ctx.SaveChanges(); } /// Seeds a Sealed deployment whose ArtifactBlob is zero-length — byte-for-byte what the /// host's ?? Array.Empty<byte>() hands downstream when the row cannot be found. private static DeploymentId SeedUnreadableDeployment(IDbContextFactory db, RevisionHash rev) => SeedDeployment(db, rev, Array.Empty()); /// Seeds a Sealed deployment carrying a real, readable artifact that keeps drv-1 (so its child /// survives the reconcile) but declares no tags for it — the driver's desired set becomes empty. private static DeploymentId SeedTaglessDriverDeployment(IDbContextFactory db, RevisionHash rev) => SeedDeployment(db, rev, JsonSerializer.SerializeToUtf8Bytes(new { RawFolders = new[] { new { RawFolderId = "rf-plant", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } }, DriverInstances = new[] { new { DriverInstanceId = "drv-1", RawFolderId = "rf-plant", Name = "Modbus", DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = true }, }, Devices = new[] { new { DeviceId = "dev-1", DriverInstanceId = "drv-1", Name = "dev1", DeviceConfig = "{}" } }, TagGroups = Array.Empty(), Tags = Array.Empty(), })); /// Seeds a Sealed deployment carrying a real, readable artifact that declares no drivers. private static DeploymentId SeedDriverlessDeployment(IDbContextFactory db, RevisionHash rev) => SeedDeployment(db, rev, JsonSerializer.SerializeToUtf8Bytes(new { RawFolders = Array.Empty(), DriverInstances = Array.Empty(), Devices = Array.Empty(), TagGroups = Array.Empty(), Tags = Array.Empty(), })); private static DeploymentId SeedDeployment( IDbContextFactory db, RevisionHash rev, byte[] artifact) { var id = DeploymentId.NewId(); using var ctx = db.CreateDbContext(); ctx.Deployments.Add(new Deployment { DeploymentId = id.Value, RevisionHash = rev.Value, Status = DeploymentStatus.Sealed, CreatedBy = "test", SealedAtUtc = DateTime.UtcNow, ArtifactBlob = artifact, }); ctx.SaveChanges(); return id; } /// Factory producing one shared for the supported type, so a /// REAL (non-stubbed) child spawns and its subscribe/unsubscribe traffic /// is observable. private sealed class SubscribingDriverFactory(string supportedType) : IDriverFactory { private readonly SubscribableStubDriver _driver = new(); /// The reference set passed to the driver's most recent SubscribeAsync call. public IReadOnlyList? LastSubscribedRefs => _driver.LastSubscribedRefs; /// Number of UnsubscribeAsync calls — non-zero means a live handle was torn down. public int UnsubscribeCount => Volatile.Read(ref _driver.UnsubscribeCount); /// public IDriver? TryCreate(string driverType, string driverInstanceId, string driverConfigJson) => string.Equals(driverType, supportedType, StringComparison.Ordinal) ? _driver : null; /// public IReadOnlyCollection SupportedTypes => new[] { supportedType }; } }