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 14b89ca9..e7065ef5 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -595,6 +595,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers private void Bootstrap() { + // Per-cluster mesh Phase 3: a FetchAndCache node NEVER reads central SQL for config, at boot + // or otherwise. It restores its served state from the LocalDb pointer instead. + if (_fetchAndCacheMode) + { + BootstrapFromCache(); + return; + } + // Read the most-recent NodeDeploymentState for this node; if it's Applied, jump // to Steady with that revision. If Applying (orphan from a crash), discard and replay. // If the DB is unreachable, fall back to Stale and start the reconnect loop. @@ -659,6 +667,60 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers } } + /// + /// Per-cluster mesh Phase 3 boot for FetchAndCache mode. Restores served state from the + /// LocalDb pointer without any central-SQL read. + /// + /// + /// + /// Distinct from (the Direct-mode SQL-unreachable fallback): + /// reading from the cache is normal operation here, not a degraded state, so + /// stays — the node can still + /// fetch new deployments. An empty or unreadable cache lands Steady-with-no-revision (the + /// first dispatch fetches), never Stale: Stale means "central SQL is down, retry it", + /// and FetchAndCache has no config SQL read to recover. + /// + /// + private void BootstrapFromCache() + { + try + { + var cached = _deploymentArtifactCache?.GetCurrentUnkeyedAsync().GetAwaiter().GetResult(); + if (cached is null) + { + _log.Info( + "DriverHost {Node}: FetchAndCache boot — the LocalDb cache is empty; entering Steady " + + "(no revision). The first dispatch will fetch this node's configuration from central.", + _localNode); + Become(Steady); + return; + } + + var deploymentId = DeploymentId.Parse(cached.DeploymentId); + var revision = RevisionHash.Parse(cached.RevisionHash); + + _currentRevision = revision; + Become(Steady); + // Restore drivers + address space + subscriptions from the cached bytes — no SQL read, no + // re-ack (the deployment is already Applied). Same in-hand-bytes apply as the Direct-mode + // cache boot, but this is the steady-state config source, not an outage fallback. + ApplyCachedArtifact(deploymentId, cached.Artifact); + + _log.Info( + "DriverHost {Node}: FetchAndCache boot — restored served state for deployment {Id} " + + "(rev {Rev}) from the LocalDb pointer.", + _localNode, deploymentId, revision); + } + catch (Exception ex) + { + _log.Warning(ex, + "DriverHost {Node}: FetchAndCache boot — failed to read the LocalDb pointer; entering " + + "Steady (no revision). The next dispatch will fetch.", + _localNode); + Become(Steady); + } + } + /// /// 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. @@ -2511,6 +2573,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers private void TryRecoverFromStale() { + // Defensive: FetchAndCache never enters Stale (BootstrapFromCache always Becomes Steady, so + // the retry-db timer that drives this is never started), and its recovery must NOT read + // central SQL. If we somehow land here, leave Steady rather than re-reading Deployments. + if (_fetchAndCacheMode) + { + Timers.Cancel("retry-db"); + Become(Steady); + return; + } + try { using var db = _dbFactory.CreateDbContext(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs new file mode 100644 index 00000000..f6d8487d --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs @@ -0,0 +1,179 @@ +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.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; + +/// +/// Per-cluster mesh Phase 3 (Task 6) — FetchAndCache bootstrap. A driver node in this mode +/// boots its served state from the LocalDb pointer, reading NO config from central SQL; an empty +/// or unreadable cache lands Steady-with-no-revision (the first dispatch fetches), NOT Stale. +/// +/// +/// Every test injects a for central SQL — reaching Steady (and, in +/// two of them, applying a dispatch) proves the boot never touched it. This is the FetchAndCache +/// counterpart to (which covers the Direct-mode +/// SQL-unreachable fallback, where RunningFromCache is TRUE — here it is normal operation, +/// so it stays FALSE). +/// +public sealed class DriverHostActorFetchAndCacheBootstrapTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-test"); + private static readonly RevisionHash CachedRev = RevisionHash.Parse(new string('c', 64)); + + private static byte[] Artifact() => + JsonSerializer.SerializeToUtf8Bytes(new { DriverInstances = Array.Empty() }); + + [Fact] + public void WithCachedArtifact_RestoresServedState_WithoutReadingCentralSql() + { + var blob = Artifact(); + var cached = new CachedDeploymentArtifact( + DeploymentId.NewId().ToString(), CachedRev.Value, blob, DateTimeOffset.UtcNow.AddHours(-1)); + var cache = new StubArtifactCache(cached); + var publishProbe = CreateTestProbe(); + + var actor = Sys.ActorOf(DriverHostActor.Props( + new ThrowingDbFactory(), TestNode, ackRouter: null, + localRoles: new HashSet { "driver" }, + opcUaPublishActor: publishProbe.Ref, + deploymentArtifactCache: cache, + fetchAndCacheMode: true, + artifactFetcher: new RecordingFetcher(Artifact()))); + + // Served state rebuilt from the CACHED bytes (no ConfigDb read — the factory throws). + var rebuild = publishProbe.ExpectMsg(TimeSpan.FromSeconds(5)); + rebuild.Artifact.ShouldBe(blob); + + var snapshot = AskDiagnostics(actor); + snapshot.CurrentRevision.ShouldBe(CachedRev); + // NORMAL operation, not the degraded Direct-mode fallback — a FetchAndCache node can still fetch. + snapshot.RunningFromCache.ShouldBeFalse(); + cache.UnkeyedReads.ShouldBe(1); + } + + [Fact] + public void WithEmptyCache_EntersSteadyNoRevision_AndTheNextDispatchFetches() + { + var cache = new StubArtifactCache(current: null); + var fetcher = new RecordingFetcher(Artifact()); + var coordinator = CreateTestProbe(); + + var actor = Sys.ActorOf(DriverHostActor.Props( + new ThrowingDbFactory(), TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: cache, + fetchAndCacheMode: true, + artifactFetcher: fetcher)); + + var snapshot = AskDiagnostics(actor); + snapshot.CurrentRevision.ShouldBeNull(); + snapshot.RunningFromCache.ShouldBeFalse(); + + // Steady, not Stale: a dispatch is serviced (Stale would ignore it, never acking) and fetches. + actor.Tell(new DispatchDeployment(DeploymentId.NewId(), CachedRev, CorrelationId.NewId())); + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + fetcher.Calls.ShouldBe(1); + } + + [Fact] + public void WithAnUnreadableCache_EntersSteadyNoRevision_NotStale() + { + var fetcher = new RecordingFetcher(Artifact()); + var coordinator = CreateTestProbe(); + + var actor = Sys.ActorOf(DriverHostActor.Props( + new ThrowingDbFactory(), TestNode, coordinator.Ref, + localRoles: new HashSet { "driver" }, + deploymentArtifactCache: new ThrowingReadCache(), + fetchAndCacheMode: true, + artifactFetcher: fetcher)); + + var snapshot = AskDiagnostics(actor); + snapshot.CurrentRevision.ShouldBeNull(); + snapshot.RunningFromCache.ShouldBeFalse(); + + // A cache-read fault is NOT Stale (Stale means "central SQL down, retry it" — there is no + // config SQL read to recover here). The node is Steady and the next dispatch fetches. + actor.Tell(new DispatchDeployment(DeploymentId.NewId(), CachedRev, CorrelationId.NewId())); + coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + fetcher.Calls.ShouldBe(1); + } + + private NodeDiagnosticsSnapshot AskDiagnostics(IActorRef actor) + { + var probe = CreateTestProbe(); + 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)); + } + + private sealed class RecordingFetcher(byte[]? bytes) : IDeploymentArtifactFetcher + { + private int _calls; + public int Calls => Volatile.Read(ref _calls); + + public Task FetchAsync(string deploymentId, string expectedRevisionHash, CancellationToken ct) + { + Interlocked.Increment(ref _calls); + return Task.FromResult(bytes); + } + } + + /// 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; + 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"); + } +}