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.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.Runtime.DeploymentCache; using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; /// /// LocalDb Phase 1 (Task 8) — booting from the node-local artifact cache when central SQL is /// unreachable. /// /// /// The behaviour that matters is asymmetric: the cache must rescue a node whose ConfigDb read /// failed, and must be completely invisible otherwise. A cache consulted on the happy path would /// be a route for stale configuration to override fresh configuration. /// public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase { private static readonly NodeId TestNode = NodeId.Parse("driver-test"); private static readonly RevisionHash CachedRev = RevisionHash.Parse(new string('c', 64)); [Fact] public void CentralUnreachableAndCacheHasArtifact_BootsFromCache() { var cached = new CachedDeploymentArtifact( DeploymentId.NewId().ToString(), CachedRev.Value, EmptyArtifact(), DateTimeOffset.UtcNow.AddHours(-1)); var cache = new StubArtifactCache(cached); var actor = Sys.ActorOf(DriverHostActor.Props( new ThrowingDbFactory(), TestNode, coordinator: null, localRoles: new HashSet { "driver" }, deploymentArtifactCache: cache)); var snapshot = AskDiagnostics(actor); snapshot.RunningFromCache.ShouldBeTrue(); snapshot.CurrentRevision.ShouldBe(CachedRev); // Positive control for the UnkeyedReads counter the "never consults the cache" tests rely // on. Without this, those ShouldBe(0) assertions would pass just as happily if the counter // were never incremented at all. cache.UnkeyedReads.ShouldBe(1); } [Fact] public void BootFromCache_HandsTheCachedArtifactToTheAddressSpaceRebuild() { // The client-facing address space rebuild must materialise from the CACHED bytes, not from a // ConfigDb read keyed by deployment id — that read is exactly what is unreachable during the // outage boot-from-cache exists to survive. Without the blob, the node logs "running from // cache", spins up its drivers, and still serves OPC UA clients an empty server. (Regression // for the defect the docker-dev live gate caught: 16 nodes on the healthy peer, 1 here.) var blob = EmptyArtifact(); var cached = new CachedDeploymentArtifact( DeploymentId.NewId().ToString(), CachedRev.Value, blob, DateTimeOffset.UtcNow.AddHours(-1)); var publishProbe = CreateTestProbe(); Sys.ActorOf(DriverHostActor.Props( new ThrowingDbFactory(), TestNode, coordinator: null, localRoles: new HashSet { "driver" }, opcUaPublishActor: publishProbe.Ref, deploymentArtifactCache: new StubArtifactCache(cached))); var rebuild = publishProbe.ExpectMsg(TimeSpan.FromSeconds(5)); rebuild.Artifact.ShouldNotBeNull(); rebuild.Artifact.ShouldBe(blob); } [Fact] public void CentralUnreachableAndCacheEmpty_StaysStaleWithNoConfiguration() { // The pre-cache behaviour, unchanged. A cache miss must not invent a configuration. var actor = Sys.ActorOf(DriverHostActor.Props( new ThrowingDbFactory(), TestNode, coordinator: null, localRoles: new HashSet { "driver" }, deploymentArtifactCache: new StubArtifactCache(current: null))); var snapshot = AskDiagnostics(actor); snapshot.RunningFromCache.ShouldBeFalse(); snapshot.CurrentRevision.ShouldBeNull(); } [Fact] public void CentralUnreachableAndNoCacheConfigured_StaysStale() { // Admin-only graphs and older harnesses pass null; behaviour is identical to a cache miss. var actor = Sys.ActorOf(DriverHostActor.Props( new ThrowingDbFactory(), TestNode, coordinator: null, localRoles: new HashSet { "driver" })); var snapshot = AskDiagnostics(actor); snapshot.RunningFromCache.ShouldBeFalse(); snapshot.CurrentRevision.ShouldBeNull(); } [Fact] public void CentralReachable_NeverConsultsTheCache() { // THE guard against stale config winning. If the cache were read on the happy path, a node // whose cache held an older revision could quietly serve it over the deployed one. var cache = new StubArtifactCache(new CachedDeploymentArtifact( DeploymentId.NewId().ToString(), CachedRev.Value, EmptyArtifact(), DateTimeOffset.UtcNow)); var actor = Sys.ActorOf(DriverHostActor.Props( NewInMemoryDbFactory(), TestNode, coordinator: null, localRoles: new HashSet { "driver" }, deploymentArtifactCache: cache)); var snapshot = AskDiagnostics(actor); snapshot.RunningFromCache.ShouldBeFalse(); cache.UnkeyedReads.ShouldBe(0); } [Fact] public void CentralReachableWithAPriorAppliedDeployment_NeverConsultsTheCache() { // The RestoreApplied path — the other way a boot can succeed against central. var db = NewInMemoryDbFactory(); var rev = RevisionHash.Parse(new string('d', 64)); var deploymentId = SeedAppliedDeployment(db, rev); var cache = new StubArtifactCache(new CachedDeploymentArtifact( deploymentId.ToString(), CachedRev.Value, EmptyArtifact(), DateTimeOffset.UtcNow)); var actor = Sys.ActorOf(DriverHostActor.Props( db, TestNode, coordinator: null, localRoles: new HashSet { "driver" }, deploymentArtifactCache: cache)); var snapshot = AskDiagnostics(actor); snapshot.RunningFromCache.ShouldBeFalse(); snapshot.CurrentRevision.ShouldBe(rev); cache.UnkeyedReads.ShouldBe(0); } [Fact] public void AThrowingCache_DegradesToStale_RatherThanFailingTheActor() { // A cache fault must leave the node exactly where it would have been without a cache. var actor = Sys.ActorOf(DriverHostActor.Props( new ThrowingDbFactory(), TestNode, coordinator: null, localRoles: new HashSet { "driver" }, deploymentArtifactCache: new ThrowingReadCache())); var snapshot = AskDiagnostics(actor); snapshot.RunningFromCache.ShouldBeFalse(); snapshot.CurrentRevision.ShouldBeNull(); } private NodeDiagnosticsSnapshot AskDiagnostics(IActorRef actor) { var probe = CreateTestProbe(); // GetDiagnostics is serviced in Steady, Applying AND Stale, so this observes the node // whichever way the boot resolved. AwaitAssert( () => { actor.Tell(new GetDiagnostics(CorrelationId.NewId()), probe.Ref); probe.ExpectMsg(TimeSpan.FromSeconds(2)); }, duration: TimeSpan.FromSeconds(5)); actor.Tell(new GetDiagnostics(CorrelationId.NewId()), probe.Ref); return probe.ExpectMsg(TimeSpan.FromSeconds(5)); } /// A syntactically valid artifact with no driver instances. private static byte[] EmptyArtifact() => JsonSerializer.SerializeToUtf8Bytes(new { DriverInstances = Array.Empty() }); private static DeploymentId SeedAppliedDeployment( IDbContextFactory db, RevisionHash rev) { 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 = EmptyArtifact(), }); ctx.NodeDeploymentStates.Add(new NodeDeploymentState { NodeId = TestNode.Value, DeploymentId = id.Value, Status = NodeDeploymentStatus.Applied, StartedAtUtc = DateTime.UtcNow, }); ctx.SaveChanges(); return id; } /// Stands in for an unreachable central SQL Server. private sealed class ThrowingDbFactory : IDbContextFactory { public OtOpcUaConfigDbContext CreateDbContext() => throw new InvalidOperationException("ConfigDb unreachable"); } private sealed class StubArtifactCache(CachedDeploymentArtifact? current) : IDeploymentArtifactCache { private int _unkeyedReads; /// How many times the boot path consulted this cache. public int UnkeyedReads => Volatile.Read(ref _unkeyedReads); public Task StoreAsync(string clusterId, string deploymentId, string revisionHash, byte[] artifact, CancellationToken ct = default) => Task.CompletedTask; public Task GetCurrentAsync( string clusterId, CancellationToken ct = default) => Task.FromResult(current); public Task GetCurrentUnkeyedAsync(CancellationToken ct = default) { Interlocked.Increment(ref _unkeyedReads); return Task.FromResult(current); } } private sealed class ThrowingReadCache : IDeploymentArtifactCache { public Task StoreAsync(string clusterId, string deploymentId, string revisionHash, byte[] artifact, CancellationToken ct = default) => Task.CompletedTask; public Task GetCurrentAsync( string clusterId, CancellationToken ct = default) => throw new InvalidOperationException("cache unreadable"); public Task GetCurrentUnkeyedAsync(CancellationToken ct = default) => throw new InvalidOperationException("cache unreadable"); } }