feat(localdb): boot from the pair-local artifact cache when central SQL is unreachable
Hooks the Bootstrap catch that previously went straight to Stale. On a cache hit the node serves its last-known-good configuration through the outage instead of coming up with an empty address space; on a miss, behaviour is byte-for-byte what it was. The read is unkeyed because ClusterId is only derivable from an artifact you already hold or from the unreachable central DB (recon D-1). Newest pointer wins, which is correct in every real topology - a node belongs to one cluster and its peer replicates that same row. Splits PushDesiredSubscriptionsFromArtifact out of PushDesiredSubscriptions: the cache path runs precisely when the ConfigDb read cannot succeed, so re-reading the artifact there would fail by definition. RunningFromCache is surfaced on NodeDiagnosticsSnapshot (optional param, existing call sites unaffected). A node running from cache looks entirely healthy from outside - full address space, live values - but its config is frozen and no deploy can reach it. It clears only on a real apply from central.
This commit is contained in:
Joseph Doherty
+232
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// LocalDb Phase 1 (Task 8) — booting from the node-local artifact cache when central SQL is
|
||||
/// unreachable.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// </remarks>
|
||||
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<string> { "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<string> { "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<string> { "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<string> { "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<string> { "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<string> { "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<NodeDiagnosticsSnapshot>(TimeSpan.FromSeconds(2));
|
||||
},
|
||||
duration: TimeSpan.FromSeconds(5));
|
||||
|
||||
actor.Tell(new GetDiagnostics(CorrelationId.NewId()), probe.Ref);
|
||||
return probe.ExpectMsg<NodeDiagnosticsSnapshot>(TimeSpan.FromSeconds(5));
|
||||
}
|
||||
|
||||
/// <summary>A syntactically valid artifact with no driver instances.</summary>
|
||||
private static byte[] EmptyArtifact()
|
||||
=> JsonSerializer.SerializeToUtf8Bytes(new { DriverInstances = Array.Empty<object>() });
|
||||
|
||||
private static DeploymentId SeedAppliedDeployment(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext> 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;
|
||||
}
|
||||
|
||||
/// <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;
|
||||
|
||||
/// <summary>How many times the boot path consulted this cache.</summary>
|
||||
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