diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs index b3932f65..c23fa6bc 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/LocalDbRegistration.cs @@ -2,6 +2,7 @@ using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using ZB.MOM.WW.LocalDb; using ZB.MOM.WW.LocalDb.Replication; +using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache; namespace ZB.MOM.WW.OtOpcUa.Host.Configuration; @@ -90,6 +91,11 @@ public static class LocalDbRegistration services.AddZbLocalDb(configuration, LocalDbSetup.OnReady); services.AddZbLocalDbReplication(configuration); + // The consumer-facing seam. WithOtOpcUaRuntimeActors resolves this optionally and threads it + // into DriverHostActor; registering the store without this leaves the cache tables present, + // replicating, and never written to. + services.AddSingleton(); + return services; } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 4a87e53c..a82fea36 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -23,6 +23,7 @@ using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms; using ZB.MOM.WW.OtOpcUa.Core.Scripting; using ZB.MOM.WW.OtOpcUa.Core.VirtualTags; using ZB.MOM.WW.OtOpcUa.OpcUaServer; +using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache; using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; using CommonsNodeId = ZB.MOM.WW.OtOpcUa.Commons.Types.NodeId; @@ -57,6 +58,19 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// Publishing interval handed to each driver's SubscribeBulk pass after an apply. private static readonly TimeSpan SubscriptionPublishingInterval = TimeSpan.FromSeconds(1); + /// + /// Cache key used when an artifact carries no cluster scoping (single-cluster or unscoped + /// deployments). A literal rather than an empty string so the row is visibly deliberate when + /// someone reads the table during an incident. + /// + private const string SingleClusterCacheKey = "__single"; + + /// + /// Node-local cache of applied deployment artifacts, or null when this node has none + /// (admin-only graphs, and tests that do not exercise it). + /// + private readonly IDeploymentArtifactCache? _deploymentArtifactCache; + private readonly IDbContextFactory _dbFactory; private readonly CommonsNodeId _localNode; private readonly IActorRef? _coordinatorOverride; @@ -339,11 +353,18 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers ScriptRootLogger? scriptRootLogger = null, IActorRef? scriptedAlarmHostOverride = null, IDriverCapabilityInvokerFactory? invokerFactory = null, - Func? driverMemberCountProvider = null) => + Func? driverMemberCountProvider = null, + IDeploymentArtifactCache? deploymentArtifactCache = null) => + // WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an + // expression tree. Six IActorRef? parameters and several interface-typed ones mean a + // mis-ordered argument is usually type-compatible and therefore compiles clean, then binds + // the wrong dependency at runtime. New parameters go LAST in all three places — this + // signature, the constructor's, and this call — and nothing else moves. Akka.Actor.Props.Create(() => new DriverHostActor( dbFactory, localNode, coordinator, driverFactory, localRoles, dependencyMux, opcUaPublishActor, healthPublisher, virtualTagEvaluator, historyWriter, virtualTagHostOverride, - loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider)); + loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider, + deploymentArtifactCache)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -372,6 +393,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// Test seam (archreview 03/S4): overrides the count of Up /// driver-role cluster members the Primary gate reads while the role is unknown. When null the /// default reads Cluster.Get(Context.System).State.Members (0 on a non-cluster ActorRefProvider). + /// Optional node-local cache of applied deployment artifacts. + /// When supplied, each successful apply stores its artifact so the node can boot from its + /// last-known-good configuration while central SQL is unreachable. Null on admin-only nodes and in + /// tests that do not exercise the cache — caching is then simply skipped. public DriverHostActor( IDbContextFactory dbFactory, CommonsNodeId localNode, @@ -388,8 +413,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers ScriptRootLogger? scriptRootLogger = null, IActorRef? scriptedAlarmHostOverride = null, IDriverCapabilityInvokerFactory? invokerFactory = null, - Func? driverMemberCountProvider = null) + Func? driverMemberCountProvider = null, + IDeploymentArtifactCache? deploymentArtifactCache = null) { + _deploymentArtifactCache = deploymentArtifactCache; _dbFactory = dbFactory; _localNode = localNode; _coordinatorOverride = coordinator; @@ -1426,7 +1453,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers try { - ReconcileDrivers(deploymentId); + var appliedBlob = ReconcileDrivers(deploymentId); _currentRevision = revision; UpsertNodeDeploymentState(deploymentId, NodeDeploymentStatus.Applied, failureReason: null); SendAck(deploymentId, ApplyAckOutcome.Applied, failureReason: null, correlation); @@ -1437,6 +1464,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers // SubscribeBulk pass: hand each driver its desired tag references so live values flow into // the just-rebuilt address space instead of staying BadWaitingForInitialData. PushDesiredSubscriptions(deploymentId); + CacheAppliedArtifact(deploymentId, revision, appliedBlob); OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair("outcome", "ack")); _log.Info("DriverHost {Node}: applied deployment {Id} (rev {Rev}, children={Count})", _localNode, deploymentId, revision, _children.Count); @@ -1464,7 +1492,20 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// configured can't materialise any of the requested /// types, this is effectively a no-op. /// - private void ReconcileDrivers(DeploymentId deploymentId) + /// + /// The artifact blob that was reconciled, or when it could not be + /// loaded at all. + /// + /// + /// Returning the blob rather than swallowing it is load-bearing for the artifact cache. This + /// method catches its own DB failures, logs a warning and returns WITHOUT rethrowing — so + /// proceeds to its success path, ACKs Applied and logs success + /// having spawned zero drivers. A cache write that trusted "the apply succeeded" would then + /// persist an empty artifact as this node's last-known-good configuration, and the node would + /// later boot from it into an empty address space. The caller distinguishes the two cases by + /// this return value. + /// + private byte[]? ReconcileDrivers(DeploymentId deploymentId) { byte[] blob; try @@ -1479,7 +1520,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers { _log.Warning(ex, "DriverHost {Node}: failed to load artifact for {Id}; skipping reconcile", _localNode, deploymentId); - return; + return null; } var specs = DeploymentArtifact.ParseDriverInstances(blob, _localNode.Value); @@ -1500,6 +1541,74 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers // sequenced AFTER ReinitializeAsync rebuilds DependencyRefs — a synchronous re-register HERE would // re-read the STALE ref set (ApplyDelta runs asynchronously on the child), so it is deliberately NOT // done inline. + + return blob; + } + + /// + /// Store a successfully applied artifact in the node-local cache, so this node can boot from + /// it while central SQL is unreachable. + /// + /// + /// + /// Never throws. By the time this runs the deployment is already recorded Applied + /// in central SQL and an Applied ACK has been sent to the coordinator. An exception + /// escaping here would unwind into 's catch and send a second, + /// contradictory Failed ACK for a deployment the fleet already believes is live. + /// + /// + /// An empty or unloadable blob is skipped, not cached. See + /// : a DB failure there degrades silently, so "the apply + /// succeeded" does not imply "a real configuration was applied". Caching an empty + /// artifact would make it this node's last-known-good and boot it into an empty address + /// space during the next outage — strictly worse than having no cache at all. + /// + /// + /// Runs synchronously on the actor thread. That matches every other DB call on this path + /// (all of which already block), and the alternative — piping the result back as a + /// self-message — would need a handler registered in Steady, Applying AND Stale or it + /// dead-letters after the Become(Steady) in the enclosing finally. + /// + /// + private void CacheAppliedArtifact(DeploymentId deploymentId, RevisionHash revision, byte[]? blob) + { + if (_deploymentArtifactCache is null) + return; + + if (blob is null || blob.Length == 0) + { + _log.Warning( + "DriverHost {Node}: not caching deployment {Id} — its artifact was empty or could not " + + "be loaded, so it is not a usable last-known-good configuration.", + _localNode, deploymentId); + return; + } + + try + { + // The node's ClusterId is carried inside the artifact (ClusterNode rows), not in config + // or on IClusterRoleInfo. A single-cluster or unscoped artifact has none, so it shares + // one sentinel key — the pair still converges on a single pointer row either way. + var scope = DeploymentArtifact.ResolveClusterScope(blob, _localNode.Value); + var clusterId = scope.Mode == ClusterFilterMode.ScopeTo && !string.IsNullOrWhiteSpace(scope.ClusterId) + ? scope.ClusterId + : SingleClusterCacheKey; + + _deploymentArtifactCache + .StoreAsync(clusterId, deploymentId.ToString(), revision.Value, blob) + .GetAwaiter() + .GetResult(); + + _log.Debug("DriverHost {Node}: cached deployment {Id} (rev {Rev}, cluster {ClusterId}, {Bytes} bytes)", + _localNode, deploymentId, revision, clusterId, blob.Length); + } + catch (Exception ex) + { + _log.Error(ex, + "DriverHost {Node}: failed to cache applied deployment {Id}. The apply itself succeeded " + + "and is unaffected; this node simply has no local fallback for that revision.", + _localNode, deploymentId); + } } /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index d3d7a4ec..0cddd870 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -15,6 +15,7 @@ using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; using ZB.MOM.WW.OtOpcUa.Core.Scripting; using ZB.MOM.WW.OtOpcUa.Core.VirtualTags; using ZB.MOM.WW.OtOpcUa.OpcUaServer; +using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache; using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; using ZB.MOM.WW.OtOpcUa.Runtime.Health; using ZB.MOM.WW.OtOpcUa.Runtime.Historian; @@ -221,6 +222,11 @@ public static class ServiceCollectionExtensions var serviceLevel = resolver.GetService() ?? NullServiceLevelPublisher.Instance; var loggerFactory = resolver.GetService() ?? NullLoggerFactory.Instance; var healthPublisher = resolver.GetService() ?? NullDriverHealthPublisher.Instance; + // Node-local deployment-artifact cache. Registered by the Host's AddOtOpcUaLocalDb on + // driver-role nodes only; deliberately left null elsewhere (admin-only graphs, test + // harnesses) rather than given a null-object, so DriverHostActor skips caching outright + // instead of pretending to cache into a sink that drops everything. + var deploymentArtifactCache = resolver.GetService(); // Root script logger backs the ScriptedAlarm host's engine + script logging. Registered in // Host DI inside the hasDriver block; may be absent in some role configs / test harnesses, // in which case the DriverHostActor gracefully skips spawning the ScriptedAlarm host. @@ -338,7 +344,8 @@ public static class ServiceCollectionExtensions historyWriter: historyWriter, loggerFactory: loggerFactory, scriptRootLogger: scriptRootLogger, - invokerFactory: invokerFactory), + invokerFactory: invokerFactory, + deploymentArtifactCache: deploymentArtifactCache), DriverHostActorName); registry.Register(driverHost); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorArtifactCacheTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorArtifactCacheTests.cs new file mode 100644 index 00000000..77f9972b --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorArtifactCacheTests.cs @@ -0,0 +1,273 @@ +using System.Text.Json; +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +using Shouldly; +using Xunit; +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.Runtime.DeploymentCache; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// LocalDb Phase 1 (Task 7) — the write half of the deployment-artifact cache. +/// +/// +/// What matters here is not that a store happens, but the conditions under which it must NOT: +/// a failed apply, and — less obviously — a "successful" apply whose artifact never actually +/// loaded. See ReconcileDrivers: it swallows its own DB failures, so apply success does +/// not imply a real configuration was applied. +/// +public sealed class DriverHostActorArtifactCacheTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-test"); + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); + + [Fact] + public void SuccessfulApply_StoresTheArtifactInTheCache() + { + var db = NewInMemoryDbFactory(); + var cache = new RecordingArtifactCache(); + var (deploymentId, artifact) = SeedDeployment(db, RevA, clusterId: null); + + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: cache)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + + AwaitAssert(() => cache.Stores.Count.ShouldBe(1), duration: TimeSpan.FromSeconds(3)); + + var store = cache.Stores[0]; + store.DeploymentId.ShouldBe(deploymentId.ToString()); + store.RevisionHash.ShouldBe(RevA.Value); + store.Artifact.ShouldBe(artifact); + } + + [Fact] + public void ApplyOfAClusterScopedArtifact_StoresUnderThatClusterId() + { + // The ClusterId is not config and not on IClusterRoleInfo — it is carried in the artifact's + // Nodes[] rows. Getting this wrong keys the pair's two nodes to different cache rows, and + // they would silently stop sharing a cache entry. + var db = NewInMemoryDbFactory(); + var cache = new RecordingArtifactCache(); + var (deploymentId, _) = SeedDeployment(db, RevA, clusterId: "SITE-A"); + + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: cache)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)); + + AwaitAssert(() => cache.Stores.Count.ShouldBe(1), duration: TimeSpan.FromSeconds(3)); + cache.Stores[0].ClusterId.ShouldBe("SITE-A"); + } + + [Fact] + public void UnscopedArtifact_StoresUnderTheSingleClusterSentinel() + { + var db = NewInMemoryDbFactory(); + var cache = new RecordingArtifactCache(); + var (deploymentId, _) = SeedDeployment(db, RevA, clusterId: null); + + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: cache)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)); + + AwaitAssert(() => cache.Stores.Count.ShouldBe(1), duration: TimeSpan.FromSeconds(3)); + cache.Stores[0].ClusterId.ShouldBe("__single"); + } + + [Fact] + public void EmptyArtifact_IsNotCached() + { + // THE case that makes this cache safe to boot from. ReconcileDrivers catches its own DB + // failures, logs a warning and returns without rethrowing — so an apply can reach its + // success path, ACK Applied, and have started zero drivers. Persisting that as + // last-known-good would boot the node into an empty address space during the next outage: + // strictly worse than having no cache at all. An empty blob is the observable form of it. + var db = NewInMemoryDbFactory(); + var cache = new RecordingArtifactCache(); + + var deploymentId = DeploymentId.NewId(); + using (var ctx = db.CreateDbContext()) + { + ctx.Deployments.Add(new Deployment + { + DeploymentId = deploymentId.Value, + RevisionHash = RevA.Value, + Status = DeploymentStatus.Sealed, + CreatedBy = "test", + SealedAtUtc = DateTime.UtcNow, + ArtifactBlob = Array.Empty(), + }); + ctx.SaveChanges(); + } + + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: cache)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + + // The apply still succeeds — an empty artifact is a legitimate no-op deployment. + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + + // But nothing is cached from it. + Thread.Sleep(500); + cache.Stores.ShouldBeEmpty(); + } + + [Fact] + public void AThrowingCache_DoesNotFailTheApply() + { + // By the time the cache runs, an Applied ACK has already gone to the coordinator. If a cache + // failure escaped, ApplyAndAck's catch would send a SECOND, contradictory Failed ACK for a + // deployment the fleet already believes is live. + var db = NewInMemoryDbFactory(); + var (deploymentId, _) = SeedDeployment(db, RevA, clusterId: null); + + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: new ThrowingArtifactCache())); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + + // Exactly one ACK, and it says Applied. A second Failed ACK arriving here would mean the + // cache failure had corrupted the deployment protocol. + coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + } + + [Fact] + public void NoCacheConfigured_AppliesNormally() + { + // Admin-only graphs and most test harnesses pass null. Caching is skipped, nothing else changes. + var db = NewInMemoryDbFactory(); + var (deploymentId, _) = SeedDeployment(db, RevA, clusterId: null); + + var coordinator = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" })); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + } + + /// + /// Seeds a sealed deployment whose artifact optionally scopes to a + /// cluster, and returns the exact bytes stored so a round-trip can be asserted. + /// + private static (DeploymentId Id, byte[] Artifact) SeedDeployment( + IDbContextFactory db, RevisionHash rev, string? clusterId) + { + object payload = clusterId is null + ? new + { + DriverInstances = Array.Empty(), + } + : new + { + DriverInstances = Array.Empty(), + // ResolveClusterScope only filters when the artifact declares MORE than one cluster; + // a single-cluster artifact is treated as unscoped. + Clusters = new object[] + { + new { ClusterId = clusterId }, + new { ClusterId = "OTHER" }, + }, + Nodes = new object[] + { + new { NodeId = TestNode.Value, ClusterId = clusterId }, + }, + }; + + var artifact = JsonSerializer.SerializeToUtf8Bytes(payload); + + 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, artifact); + } + + /// Records every store so the test can assert on what was cached. + private sealed class RecordingArtifactCache : IDeploymentArtifactCache + { + private readonly Lock _gate = new(); + private readonly List _stores = []; + + public IReadOnlyList Stores + { + get { lock (_gate) { return _stores.ToArray(); } } + } + + public Task StoreAsync(string clusterId, string deploymentId, string revisionHash, + byte[] artifact, CancellationToken ct = default) + { + lock (_gate) + { + _stores.Add(new StoreCall(clusterId, deploymentId, revisionHash, artifact)); + } + + return Task.CompletedTask; + } + + public Task GetCurrentAsync( + string clusterId, CancellationToken ct = default) + => Task.FromResult(null); + + public Task GetCurrentUnkeyedAsync(CancellationToken ct = default) + => Task.FromResult(null); + + internal sealed record StoreCall( + string ClusterId, string DeploymentId, string RevisionHash, byte[] Artifact); + } + + /// Fails every store, to prove a cache fault cannot reach the deployment protocol. + private sealed class ThrowingArtifactCache : IDeploymentArtifactCache + { + public Task StoreAsync(string clusterId, string deploymentId, string revisionHash, + byte[] artifact, CancellationToken ct = default) + => throw new InvalidOperationException("disk full"); + + public Task GetCurrentAsync( + string clusterId, CancellationToken ct = default) + => Task.FromResult(null); + + public Task GetCurrentUnkeyedAsync(CancellationToken ct = default) + => Task.FromResult(null); + } +}