feat(mesh): FetchAndCache bootstrap from the LocalDb pointer; no driver-side config SQL read
Phase 3 Task 6. In FetchAndCache mode Bootstrap() branches to BootstrapFromCache(), which restores served state from the LocalDb pointer (GetCurrentUnkeyedAsync → set _currentRevision → ApplyCachedArtifact from the cached bytes) with NO central-SQL read. Distinct from the Direct-mode TryBootFromCache fallback: reading the cache is NORMAL operation here so _isRunningFromCache stays false (the node can still fetch new deploys). An empty OR unreadable cache lands Steady-with-no-revision (the first dispatch fetches), never Stale — Stale means 'central SQL down, retry it', and FetchAndCache has no config SQL read to recover. TryRecoverFromStale gains a defensive FetchAndCache guard (it is unreachable in that mode, since the retry-db timer only starts in Stale). Proven with a ThrowingDbFactory in every test: reaching Steady + applying a dispatch proves the boot never touched central SQL. Sabotaging the mode branch (fall through to the SQL read) reddens all three — test A flips RunningFromCache; B/C enter Stale, which ignores the dispatch. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
+179
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Per-cluster mesh Phase 3 (Task 6) — <c>FetchAndCache</c> 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.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Every test injects a <see cref="ThrowingDbFactory"/> 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 <see cref="DriverHostActorBootFromCacheTests"/> (which covers the Direct-mode
|
||||
/// SQL-unreachable fallback, where <c>RunningFromCache</c> is TRUE — here it is normal operation,
|
||||
/// so it stays FALSE).
|
||||
/// </remarks>
|
||||
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<object>() });
|
||||
|
||||
[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<string> { "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<OpcUaPublishActor.RebuildAddressSpace>(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<string> { "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<ApplyAck>(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<string> { "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<ApplyAck>(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<NodeDiagnosticsSnapshot>(TimeSpan.FromSeconds(2));
|
||||
},
|
||||
duration: TimeSpan.FromSeconds(5));
|
||||
|
||||
actor.Tell(new GetDiagnostics(CorrelationId.NewId()), probe.Ref);
|
||||
return probe.ExpectMsg<NodeDiagnosticsSnapshot>(TimeSpan.FromSeconds(5));
|
||||
}
|
||||
|
||||
private sealed class RecordingFetcher(byte[]? bytes) : IDeploymentArtifactFetcher
|
||||
{
|
||||
private int _calls;
|
||||
public int Calls => Volatile.Read(ref _calls);
|
||||
|
||||
public Task<byte[]?> FetchAsync(string deploymentId, string expectedRevisionHash, CancellationToken ct)
|
||||
{
|
||||
Interlocked.Increment(ref _calls);
|
||||
return Task.FromResult(bytes);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Stands in for an unreachable central SQL Server.</summary>
|
||||
private sealed class ThrowingDbFactory : IDbContextFactory<OtOpcUaConfigDbContext>
|
||||
{
|
||||
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<CachedDeploymentArtifact?> GetCurrentAsync(string clusterId, CancellationToken ct = default)
|
||||
=> Task.FromResult(current);
|
||||
|
||||
public Task<CachedDeploymentArtifact?> 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<CachedDeploymentArtifact?> GetCurrentAsync(string clusterId, CancellationToken ct = default)
|
||||
=> throw new InvalidOperationException("cache unreadable");
|
||||
|
||||
public Task<CachedDeploymentArtifact?> GetCurrentUnkeyedAsync(CancellationToken ct = default)
|
||||
=> throw new InvalidOperationException("cache unreadable");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user