diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Interfaces/NodeDiagnosticsSnapshot.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Interfaces/NodeDiagnosticsSnapshot.cs index e31b62ff..a2fafebd 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Interfaces/NodeDiagnosticsSnapshot.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Interfaces/NodeDiagnosticsSnapshot.cs @@ -14,8 +14,16 @@ public sealed record DriverInstanceDiagnostics( /// Per-node diagnostics returned by IFleetDiagnosticsClient. Populated by the node's /// local DriverHostActor via a request/response over Akka. /// +/// +/// True when the node booted its configuration from the node-local artifact cache because the +/// central ConfigDb was unreachable. Such a node looks entirely healthy — full address space, +/// live values — but its configuration is frozen and no deployment can reach it, so the state +/// needs to be explicitly visible rather than inferred. Defaults to false, which is also what +/// every node that booted normally reports. +/// public sealed record NodeDiagnosticsSnapshot( NodeId NodeId, RevisionHash? CurrentRevision, IReadOnlyList Drivers, - DateTime AsOfUtc); + DateTime AsOfUtc, + bool RunningFromCache = false); 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 a82fea36..8b1df887 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -71,6 +71,18 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// private readonly IDeploymentArtifactCache? _deploymentArtifactCache; + /// + /// True once this node has booted a cached artifact because central SQL was unreachable. + /// + /// + /// Surfaced on because a node running from cache looks + /// completely healthy from the outside — it serves a full address space with live values. + /// The difference is that its configuration is frozen at whatever the cache held, and no + /// deployment can reach it. Without an explicit signal that is invisible until someone + /// wonders why a deploy "succeeded" everywhere but did not take effect here. + /// + private bool _isRunningFromCache; + private readonly IDbContextFactory _dbFactory; private readonly CommonsNodeId _localNode; private readonly IActorRef? _coordinatorOverride; @@ -587,10 +599,118 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers catch (Exception ex) { _log.Warning(ex, "DriverHost {Node}: ConfigDb unreachable on bootstrap; entering Stale", _localNode); - Become(Stale); + + // Central is unreachable, so try the node-local cache before giving up. On a hit this + // node serves its last-known-good configuration through the outage instead of coming up + // with an empty address space. On a miss, behaviour is exactly what it always was. + if (!TryBootFromCache()) + Become(Stale); } } + /// + /// Last-resort boot path: apply the artifact cached by a previous successful deploy (or + /// replicated from this node's pair peer) when central SQL cannot be reached. + /// + /// + /// when a cached artifact was applied and the actor is now Steady; + /// when the caller should fall through to Stale. + /// + /// + /// + /// The read is unkeyed. The cache is keyed by ClusterId so a pair shares one + /// entry, but ClusterId is only derivable from an artifact you already hold or from the + /// central DB — and at this seam we have neither. So the newest pointer row wins, which + /// is correct in every real topology because a node belongs to one cluster and its peer + /// replicates that same cluster's row. A node re-homed between clusters is the only + /// ambiguous case, and the cache logs a warning naming both. + /// + /// + /// This does not make the node current. _currentRevision is set from the + /// cached artifact, so a subsequent dispatch of that same revision correctly no-ops, + /// while any NEW revision still requires central — the cache holds the past, not the + /// future. Dispatch handling is unchanged. + /// + /// + /// Never throws: a fault here must degrade to Stale, which is exactly where the node + /// would have been without a cache at all. + /// + /// + private bool TryBootFromCache() + { + if (_deploymentArtifactCache is null) + return false; + + try + { + var cached = _deploymentArtifactCache.GetCurrentUnkeyedAsync().GetAwaiter().GetResult(); + if (cached is null) + { + _log.Info( + "DriverHost {Node}: no cached deployment available; entering Stale with no configuration.", + _localNode); + return false; + } + + var deploymentId = DeploymentId.Parse(cached.DeploymentId); + var revision = RevisionHash.Parse(cached.RevisionHash); + + // Steady, not Stale: this node is serving a real configuration. The retry-db timer that + // Stale would have started does not run here, so recovery rides on the next dispatch — + // matching how a normally-booted node behaves. + _currentRevision = revision; + Become(Steady); + + ApplyCachedArtifact(deploymentId, cached.Artifact); + + _isRunningFromCache = true; + _log.Warning( + "DriverHost {Node}: RUNNING FROM CACHE — central ConfigDb is unreachable, so this node " + + "booted deployment {Id} (rev {Rev}) cached at {CachedAtUtc:o} from its local database. " + + "Configuration changes cannot be applied until the ConfigDb is reachable again.", + _localNode, deploymentId, revision, cached.AppliedAtUtc); + + return true; + } + catch (Exception ex) + { + _log.Error(ex, + "DriverHost {Node}: failed to boot from the local deployment cache; falling back to Stale.", + _localNode); + return false; + } + } + + /// + /// Applies an artifact already in hand — no ConfigDb read, no ACK. + /// + /// + /// Mirrors , but sources the artifact from the cache rather than + /// re-reading it from a database that is by definition unreachable here. It also skips + /// UpsertNodeDeploymentState and SendAck for the same reason: both write to or + /// depend on central. + /// + private void ApplyCachedArtifact(DeploymentId deploymentId, byte[] blob) + { + var correlation = CorrelationId.NewId(); + + var specs = DeploymentArtifact.ParseDriverInstances(blob, _localNode.Value); + var snapshots = _children.ToDictionary( + kv => kv.Key, + kv => new DriverChildSnapshot(kv.Value.DriverType, kv.Value.LastConfigJson, kv.Value.ResilienceConfig), + StringComparer.Ordinal); + var plan = DriverSpawnPlanner.Compute(snapshots, specs); + + foreach (var id in plan.ToStop) StopChild(id); + foreach (var spec in plan.ToApplyDelta) ApplyChildDelta(spec); + foreach (var spec in plan.ToSpawn) SpawnChild(spec); + + _opcUaPublishActor?.Tell( + new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RebuildAddressSpace(correlation, deploymentId)); + + PushDesiredSubscriptionsFromArtifact(deploymentId, blob); + } + private void Steady() { Receive(HandleDispatchFromSteady); @@ -1420,7 +1540,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers NodeId: _localNode, CurrentRevision: _currentRevision, Drivers: drivers, - AsOfUtc: DateTime.UtcNow); + AsOfUtc: DateTime.UtcNow, + RunningFromCache: _isRunningFromCache); Sender.Tell(snapshot); } @@ -1465,6 +1586,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers // the just-rebuilt address space instead of staying BadWaitingForInitialData. PushDesiredSubscriptions(deploymentId); CacheAppliedArtifact(deploymentId, revision, appliedBlob); + // Reaching here means the artifact came from central, so the node is no longer serving a + // cache-sourced configuration. Note this only clears on a real apply: a dispatch of the + // revision already booted from cache short-circuits in HandleDispatchFromSteady without + // touching the ConfigDb, and the flag correctly stays set. + _isRunningFromCache = false; OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair("outcome", "ack")); _log.Info("DriverHost {Node}: applied deployment {Id} (rev {Rev}, children={Count})", _localNode, deploymentId, revision, _children.Count); @@ -1661,6 +1787,19 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers return; } + PushDesiredSubscriptionsFromArtifact(deploymentId, blob); + } + + /// + /// The artifact-processing half of , split out so the + /// boot-from-cache path can drive it with bytes already in hand. + /// + /// + /// Separated because the cache path runs precisely when the ConfigDb read above cannot + /// succeed — re-reading the artifact there would fail by definition. + /// + private void PushDesiredSubscriptionsFromArtifact(DeploymentId deploymentId, byte[] blob) + { AddressSpaceComposition composition; try { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs new file mode 100644 index 00000000..1354a30e --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs @@ -0,0 +1,232 @@ +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.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 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"); + } +}