feat(localdb): DriverHostActor stores applied artifacts in the pair-local cache
New Props/ctor parameter appended LAST at all three positions - the forwarding list inside Props.Create is positional and compiles into an expression tree, so a mis-ordered argument is usually type-compatible and binds the wrong dependency at runtime. All 26 existing test construction sites use named arguments and needed no change. ReconcileDrivers now returns the blob it loaded. It catches its own DB failures and returns without rethrowing, so ApplyAndAck can reach its success path having spawned zero drivers; caching on 'the apply succeeded' would persist an empty artifact as last-known-good and boot the node into an empty address space during the next outage. Verified: weakening the guard to null-only turns EmptyArtifact_IsNotCached red. The store runs in its own try/catch because an Applied ACK has already been sent by then - an escaping exception would land in ApplyAndAck's catch and emit a second, contradictory Failed ACK for a deployment the fleet believes is live.
This commit is contained in:
Joseph Doherty
+273
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// LocalDb Phase 1 (Task 7) — the write half of the deployment-artifact cache.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 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 <c>ReconcileDrivers</c>: it swallows its own DB failures, so apply success does
|
||||
/// not imply a real configuration was applied.
|
||||
/// </remarks>
|
||||
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<string> { "driver" },
|
||||
deploymentArtifactCache: cache));
|
||||
|
||||
actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
|
||||
coordinator.ExpectMsg<ApplyAck>(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<string> { "driver" },
|
||||
deploymentArtifactCache: cache));
|
||||
|
||||
actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
|
||||
coordinator.ExpectMsg<ApplyAck>(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<string> { "driver" },
|
||||
deploymentArtifactCache: cache));
|
||||
|
||||
actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
|
||||
coordinator.ExpectMsg<ApplyAck>(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<byte>(),
|
||||
});
|
||||
ctx.SaveChanges();
|
||||
}
|
||||
|
||||
var coordinator = CreateTestProbe();
|
||||
var actor = Sys.ActorOf(DriverHostActor.Props(
|
||||
db, TestNode, coordinator.Ref,
|
||||
localRoles: new HashSet<string> { "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<ApplyAck>(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<string> { "driver" },
|
||||
deploymentArtifactCache: new ThrowingArtifactCache()));
|
||||
|
||||
actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
|
||||
|
||||
coordinator.ExpectMsg<ApplyAck>(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<string> { "driver" }));
|
||||
|
||||
actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
|
||||
|
||||
coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Seeds a sealed deployment whose artifact optionally scopes <see cref="TestNode"/> to a
|
||||
/// cluster, and returns the exact bytes stored so a round-trip can be asserted.
|
||||
/// </summary>
|
||||
private static (DeploymentId Id, byte[] Artifact) SeedDeployment(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext> db, RevisionHash rev, string? clusterId)
|
||||
{
|
||||
object payload = clusterId is null
|
||||
? new
|
||||
{
|
||||
DriverInstances = Array.Empty<object>(),
|
||||
}
|
||||
: new
|
||||
{
|
||||
DriverInstances = Array.Empty<object>(),
|
||||
// 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);
|
||||
}
|
||||
|
||||
/// <summary>Records every store so the test can assert on what was cached.</summary>
|
||||
private sealed class RecordingArtifactCache : IDeploymentArtifactCache
|
||||
{
|
||||
private readonly Lock _gate = new();
|
||||
private readonly List<StoreCall> _stores = [];
|
||||
|
||||
public IReadOnlyList<StoreCall> 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<CachedDeploymentArtifact?> GetCurrentAsync(
|
||||
string clusterId, CancellationToken ct = default)
|
||||
=> Task.FromResult<CachedDeploymentArtifact?>(null);
|
||||
|
||||
public Task<CachedDeploymentArtifact?> GetCurrentUnkeyedAsync(CancellationToken ct = default)
|
||||
=> Task.FromResult<CachedDeploymentArtifact?>(null);
|
||||
|
||||
internal sealed record StoreCall(
|
||||
string ClusterId, string DeploymentId, string RevisionHash, byte[] Artifact);
|
||||
}
|
||||
|
||||
/// <summary>Fails every store, to prove a cache fault cannot reach the deployment protocol.</summary>
|
||||
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<CachedDeploymentArtifact?> GetCurrentAsync(
|
||||
string clusterId, CancellationToken ct = default)
|
||||
=> Task.FromResult<CachedDeploymentArtifact?>(null);
|
||||
|
||||
public Task<CachedDeploymentArtifact?> GetCurrentUnkeyedAsync(CancellationToken ct = default)
|
||||
=> Task.FromResult<CachedDeploymentArtifact?>(null);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user