Merge rows 78+72: DeploymentManager singleton instance sweep — starts replication-seeded enabled instances and converges stale revisions through the real HandleDeploy path; one repair attempt per (instance, revision)

This commit is contained in:
Joseph Doherty
2026-08-16 04:17:45 -04:00
7 changed files with 1186 additions and 16 deletions
@@ -0,0 +1,549 @@
using System.Text.Json;
using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.TestSupport;
using ZB.MOM.WW.ScadaBridge.TestSupport;
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors;
/// <summary>
/// Regression tests for deferred register rows <b>78</b> and <b>72</b> — the two ways the site
/// runtime silently diverges from its own durable <c>deployed_configurations</c> rows.
///
/// <para>
/// <b>Row 78 (availability).</b> Instance Actors were created from three paths: the boot-time
/// SQLite load, an explicit deploy/enable command, and (row 67) the startup reconcile's
/// <see cref="EnsureInstanceStartedCommand"/> for configs the reconcile ITSELF fetched. A config
/// that arrives by LocalDb CDC <i>replication</i> hits none of them: the sync applier writes the
/// row straight into the file, AFTER the boot load has read storage and invisibly to the
/// reconcile, which correctly reports no gap because central's view of the site is right. Proven
/// live 2026-08-15 — a fresh site-a-a was seeded from its peer within ~5 s of boot and its
/// enabled instances sat dead, with their configs present, until another restart.
/// </para>
///
/// <para>
/// <b>Row 72 (silent divergence).</b> The reconcile refreshes a STALE row via
/// <c>StoreDeployedConfigIfNewerAsync</c>, but the running Instance Actor keeps serving the OLD
/// compiled config until the next restart or redeploy, and nothing surfaced
/// "stored revision != running revision".
/// </para>
///
/// <para>
/// Both are closed by one periodic sweep owned by the Deployment Manager singleton. Design note:
/// <c>docs/plans/2026-08-15-instance-sweep-self-heal-design.md</c>.
/// </para>
/// </summary>
public class InstanceSweepSelfHealTests : TestKit, IDisposable
{
private readonly SiteStorageService _storage;
private readonly ScriptCompilationService _compilationService;
private readonly SharedScriptLibrary _sharedScriptLibrary;
private readonly TestLocalDb _localDb;
public InstanceSweepSelfHealTests()
{
_localDb = TestLocalDb.CreateTemp("instance-sweep-test");
_storage = new SiteStorageService(_localDb.Db, NullLogger<SiteStorageService>.Instance);
_storage.InitializeAsync().GetAwaiter().GetResult();
_compilationService = new ScriptCompilationService(
NullLogger<ScriptCompilationService>.Instance);
_sharedScriptLibrary = new SharedScriptLibrary(
_compilationService, NullLogger<SharedScriptLibrary>.Instance);
}
void IDisposable.Dispose()
{
// Shut the actors down BEFORE the ILocalDb they write through is disposed.
Shutdown();
var path = _localDb.Path;
_localDb.Dispose();
TestLocalDb.DeleteFiles(path);
}
/// <summary>
/// Sweep interval used by these tests. One second is the shortest the (whole-second) option
/// allows; every assertion below therefore budgets several seconds so it is waiting on the
/// behaviour, not on the clock.
/// </summary>
private const int SweepSeconds = 1;
/// <summary>Long enough for at least three sweep ticks to have come and gone.</summary>
private static readonly TimeSpan SeveralSweeps = TimeSpan.FromSeconds(4);
/// <summary>
/// Starts a Deployment Manager and blocks until its boot-time config load has PROVABLY read
/// storage (its first instance-count report). Load-bearing for every test here: a row written
/// before that read is picked up by the boot load itself, which is precisely the path row 78
/// is about NOT having. Without this gate a test would silently prove the boot load works.
/// </summary>
private IActorRef CreateBootedDeploymentManager(
FakeSiteEventLogger? siteLog = null,
int sweepSeconds = SweepSeconds)
{
var health = new StartupProbe();
var dm = ActorOf(Props.Create(() => new DeploymentManagerActor(
_storage,
_compilationService,
_sharedScriptLibrary,
null,
new SiteRuntimeOptions { InstanceSweepIntervalSeconds = sweepSeconds },
NullLogger<DeploymentManagerActor>.Instance,
dclManager: null,
healthCollector: health,
serviceProvider: siteLog == null ? null : new SingleServiceProvider(siteLog),
loggerFactory: null,
configFetcher: null)));
AwaitCondition(() => health.ReportCount > 0, TimeSpan.FromSeconds(20),
TimeSpan.FromMilliseconds(25));
return dm;
}
private static string MakeConfigJson(string instanceName, string attributeValue = "42") =>
JsonSerializer.Serialize(new FlattenedConfiguration
{
InstanceUniqueName = instanceName,
Attributes =
[
new ResolvedAttribute
{
CanonicalName = "TestAttr", Value = attributeValue, DataType = "Int32"
}
]
});
/// <summary>
/// A config carrying a real script, so its compile gate / compile warm takes long enough for
/// the deploy pipeline to hold the instance across a sweep tick.
/// </summary>
private static string MakeScriptedConfigJson(string instanceName, string scriptCode) =>
JsonSerializer.Serialize(new FlattenedConfiguration
{
InstanceUniqueName = instanceName,
Attributes =
[
new ResolvedAttribute
{
CanonicalName = "TestAttr", Value = "42", DataType = "Int32"
}
],
Scripts =
[
new ResolvedScript
{
CanonicalName = "Worker", Code = scriptCode, TriggerType = "Call"
}
]
});
/// <summary>
/// Asks the Deployment Manager for a debug snapshot — the exact probe that answered
/// <c>instanceNotFound</c> on the live rig.
/// </summary>
private async Task<DebugViewSnapshot> SnapshotAsync(IActorRef dm, string instanceName) =>
await dm.Ask<DebugViewSnapshot>(
new SubscribeDebugViewRequest(instanceName, Guid.NewGuid().ToString()),
TimeSpan.FromSeconds(5));
private async Task<bool> InstanceIsRunningAsync(IActorRef dm, string instanceName) =>
!(await SnapshotAsync(dm, instanceName)).InstanceNotFound;
private async Task<IActorRef> ResolveInstanceChildAsync(IActorRef dm, string instanceName) =>
await Sys.ActorSelection(dm.Path / instanceName).ResolveOne(TimeSpan.FromSeconds(5));
/// <summary>
/// Site events announcing a sweep REPAIR (a start or a stale-revision restart). The terminal
/// "self-heal failed" event is deliberately excluded — it reports the outcome of a repair
/// already counted here, so counting both would double-count a single attempt.
/// </summary>
private static IReadOnlyList<FakeSiteEventLogger.Entry> SelfHealEvents(
FakeSiteEventLogger log, string instanceName) =>
log.OfType("deployment")
.Where(e => e.InstanceId == instanceName &&
e.Message.Contains("self-heal", StringComparison.OrdinalIgnoreCase) &&
!e.Message.Contains("self-heal failed", StringComparison.OrdinalIgnoreCase))
.ToArray();
/// <summary>
/// Deploys an instance the ordinary way and blocks until it is running, so a test can start
/// from a known-good runtime state.
/// </summary>
private async Task DeployAndAwaitAsync(
IActorRef dm, string instanceName, string revision, string configJson)
{
dm.Tell(new DeployInstanceCommand(
$"dep-{revision}", instanceName, revision, configJson, "admin", DateTimeOffset.UtcNow));
var response = ExpectMsg<DeploymentStatusResponse>(TimeSpan.FromSeconds(15));
Assert.Equal(DeploymentStatus.Success, response.Status);
await AwaitAssertAsync(async () =>
Assert.True(await InstanceIsRunningAsync(dm, instanceName)), TimeSpan.FromSeconds(10));
}
// ── Row 78: a config that arrives with no deploy, no enable and no reconcile ──
/// <summary>
/// THE row-78 reproduction. The node boots with an empty deployed-config store (so the
/// boot-time load starts nothing), and the config for an ENABLED instance then appears in
/// storage on its own — exactly what LocalDb's CDC sync applier does when it seeds a fresh
/// node from its peer. No <see cref="DeployInstanceCommand"/>, no
/// <see cref="EnsureInstanceStartedCommand"/>, no reconcile actor: nothing but the row.
///
/// <para>
/// <b>Fails before the fix</b> at the final assert — the row is in storage and the debug
/// snapshot still answers <c>InstanceNotFound = true</c>, the live symptom verbatim.
/// </para>
/// </summary>
[Fact]
public async Task Sweep_StartsEnabledInstance_WhoseConfigArrivedWithoutADeployOrReconcile()
{
const string instanceName = "ReplicatedPump";
var dm = CreateBootedDeploymentManager();
// Baseline: the node genuinely booted with nothing.
await AwaitAssertAsync(async () =>
Assert.False(await InstanceIsRunningAsync(dm, instanceName)), TimeSpan.FromSeconds(5));
// Replication seeds the row. This is a bare storage write on purpose — it is the only
// side effect the CDC applier has, and it is invisible to every other start path.
await _storage.StoreDeployedConfigAsync(
instanceName, MakeConfigJson(instanceName), "dep-replicated", "rev-replicated", true);
await AwaitAssertAsync(async () =>
Assert.True(
await InstanceIsRunningAsync(dm, instanceName),
"A replication-seeded config for an enabled instance left no running Instance "
+ "Actor — the instance would stay dead until the next restart (register row 78)"),
TimeSpan.FromSeconds(20));
}
/// <summary>
/// The heal must be attributable: it emits the same Warning <c>deployment</c> site event
/// pattern as row 67, with the trigger named so an operator can tell a sweep heal from a
/// startup-reconcile heal.
/// </summary>
[Fact]
public async Task SweepStartedInstance_EmitsAWarningSiteEvent_NamingTheSweepTrigger()
{
const string instanceName = "SurfacedSweepPump";
var siteLog = new FakeSiteEventLogger();
var dm = CreateBootedDeploymentManager(siteLog);
await _storage.StoreDeployedConfigAsync(
instanceName, MakeConfigJson(instanceName), "dep-s", "rev-s", true);
await AwaitAssertAsync(() =>
{
Assert.Contains(siteLog.OfType("deployment"), e =>
e.InstanceId == instanceName &&
e.Severity == "Warning" &&
e.Source == "DeploymentManagerActor" &&
e.Message.Contains("deployment-manager sweep", StringComparison.OrdinalIgnoreCase) &&
e.Message.Contains("self-heal", StringComparison.OrdinalIgnoreCase));
return Task.CompletedTask;
}, TimeSpan.FromSeconds(20));
}
// ── Row 78 negative guards ──
/// <summary>
/// A DISABLED instance legitimately has a config row and no running actor. The sweep must
/// never start it — that would silently undo an operator's disable.
/// </summary>
[Fact]
public async Task Sweep_NeverStarts_ADisabledInstance()
{
const string instanceName = "DisabledSweepPump";
var siteLog = new FakeSiteEventLogger();
var dm = CreateBootedDeploymentManager(siteLog);
await _storage.StoreDeployedConfigAsync(
instanceName, MakeConfigJson(instanceName), "dep-d", "rev-d", false);
await Task.Delay(SeveralSweeps);
Assert.False(await InstanceIsRunningAsync(dm, instanceName));
Assert.Empty(SelfHealEvents(siteLog, instanceName));
}
/// <summary>
/// Idempotence: many sweeps, one start. The sweep runs forever on a live node, so a repair
/// that re-fired every tick would be a restart loop plus an unbounded stream of Warning site
/// events.
/// </summary>
[Fact]
public async Task Sweep_StartsTheInstanceExactlyOnce_AcrossManySweeps()
{
const string instanceName = "IdempotentSweepPump";
var siteLog = new FakeSiteEventLogger();
var dm = CreateBootedDeploymentManager(siteLog);
await _storage.StoreDeployedConfigAsync(
instanceName, MakeConfigJson(instanceName), "dep-i", "rev-i", true);
await AwaitAssertAsync(async () =>
Assert.True(await InstanceIsRunningAsync(dm, instanceName)), TimeSpan.FromSeconds(20));
var afterFirstHeal = await ResolveInstanceChildAsync(dm, instanceName);
// Let several more sweeps run over the now-healthy state.
await Task.Delay(SeveralSweeps);
Assert.Equal(afterFirstHeal, await ResolveInstanceChildAsync(dm, instanceName));
Assert.Single(SelfHealEvents(siteLog, instanceName));
}
/// <summary>
/// A sweep that finds nothing wrong must be completely silent — no site events at all. This
/// runs on every tick of every healthy site, forever.
/// </summary>
[Fact]
public async Task Sweep_IsSilent_WhenThereIsNothingToHeal()
{
const string instanceName = "HealthySweepPump";
var siteLog = new FakeSiteEventLogger();
var dm = CreateBootedDeploymentManager(siteLog);
await DeployAndAwaitAsync(dm, instanceName, "rev-1", MakeConfigJson(instanceName));
var eventsAfterDeploy = siteLog.Entries.Count;
await Task.Delay(SeveralSweeps);
Assert.Equal(eventsAfterDeploy, siteLog.Entries.Count);
Assert.True(await InstanceIsRunningAsync(dm, instanceName));
}
/// <summary>
/// The sweep is opt-out: <c>InstanceSweepIntervalSeconds = 0</c> arms no timer at all, so a
/// site that wants the pre-fix behaviour keeps it (and the option's disable path is proven,
/// not merely validated).
/// </summary>
[Fact]
public async Task Sweep_DoesNothing_WhenTheIntervalIsZero()
{
const string instanceName = "NoSweepPump";
var siteLog = new FakeSiteEventLogger();
var dm = CreateBootedDeploymentManager(siteLog, sweepSeconds: 0);
await _storage.StoreDeployedConfigAsync(
instanceName, MakeConfigJson(instanceName), "dep-z", "rev-z", true);
await Task.Delay(SeveralSweeps);
Assert.False(await InstanceIsRunningAsync(dm, instanceName));
Assert.Empty(SelfHealEvents(siteLog, instanceName));
}
/// <summary>
/// The sweep must defer to any deploy pipeline that owns the instance. A redeploy passes
/// through a window where the predecessor has been stopped and removed from routing but its
/// replacement does not exist yet — the sweep would read that as "enabled config, no actor"
/// and try to create a child whose actor name is still registered.
///
/// <para>
/// Sweep ticks are fired continuously across the whole redeploy so the window is genuinely
/// sampled; the assertion is that the redeploy succeeded untouched and the sweep announced
/// nothing.
/// </para>
/// </summary>
[Fact]
public async Task Sweep_DefersToAnInFlightRedeploy()
{
const string instanceName = "RedeployingPump";
var siteLog = new FakeSiteEventLogger();
var dm = CreateBootedDeploymentManager(siteLog);
await DeployAndAwaitAsync(
dm, instanceName, "rev-1",
MakeScriptedConfigJson(instanceName, "var a = 1; var b = a + 1; return b;"));
using var ticking = new CancellationTokenSource();
var ticker = Task.Run(async () =>
{
while (!ticking.IsCancellationRequested)
{
dm.Tell(DeploymentManagerActor.InstanceSweepTick.Instance);
await Task.Delay(5);
}
});
dm.Tell(new DeployInstanceCommand(
"dep-rev-2", instanceName, "rev-2",
MakeScriptedConfigJson(instanceName, "var x = 2; var y = x * 3; return y;"),
"admin", DateTimeOffset.UtcNow));
var response = ExpectMsg<DeploymentStatusResponse>(TimeSpan.FromSeconds(20));
await ticking.CancelAsync();
await ticker;
Assert.Equal(DeploymentStatus.Success, response.Status);
Assert.True(await InstanceIsRunningAsync(dm, instanceName));
Assert.Empty(SelfHealEvents(siteLog, instanceName));
}
// ── Row 72: a running actor serving a revision the store has replaced ──
/// <summary>
/// THE row-72 reproduction. An instance is deployed and running at <c>rev-1</c>; the durable
/// row is then refreshed to <c>rev-2</c> out of band — what
/// <c>StoreDeployedConfigIfNewerAsync</c> does when the startup reconcile finds a STALE
/// config, and what a replicated deploy does on the standby.
///
/// <para>
/// <b>Fails before the fix</b>: the actor keeps serving <c>rev-1</c>'s config indefinitely
/// (the snapshot still shows the old attribute value) and nothing anywhere says the stored
/// and running revisions differ.
/// </para>
/// </summary>
[Fact]
public async Task Sweep_RestartsInstance_ServingAStaleRevision()
{
const string instanceName = "StalePump";
var siteLog = new FakeSiteEventLogger();
var dm = CreateBootedDeploymentManager(siteLog);
await DeployAndAwaitAsync(
dm, instanceName, "rev-1", MakeConfigJson(instanceName, attributeValue: "111"));
var beforeHeal = await ResolveInstanceChildAsync(dm, instanceName);
// The reconcile refreshes the stale row. Storage only; the running actor is untouched.
await _storage.StoreDeployedConfigAsync(
instanceName, MakeConfigJson(instanceName, attributeValue: "222"),
"dep-rev-2", "rev-2", true);
// The runtime must converge on the stored config...
await AwaitAssertAsync(async () =>
{
var snapshot = await SnapshotAsync(dm, instanceName);
Assert.False(snapshot.InstanceNotFound);
Assert.Contains(snapshot.AttributeValues, a =>
a.AttributeName == "TestAttr" && a.Value?.ToString() == "222");
}, TimeSpan.FromSeconds(25));
// ...through a genuine restart, not an in-place mutation.
Assert.NotEqual(beforeHeal, await ResolveInstanceChildAsync(dm, instanceName));
// ...and the divergence must be surfaced, naming BOTH revisions.
Assert.Contains(SelfHealEvents(siteLog, instanceName), e =>
e.Severity == "Warning" &&
e.Message.Contains("rev-1", StringComparison.Ordinal) &&
e.Message.Contains("rev-2", StringComparison.Ordinal));
}
/// <summary>
/// The stale-revision restart must not fire when the stored and running revisions agree —
/// otherwise every healthy instance on the site would be restarted once a minute.
/// </summary>
[Fact]
public async Task Sweep_DoesNotRestart_WhenStoredAndRunningRevisionsMatch()
{
const string instanceName = "MatchingRevisionPump";
var siteLog = new FakeSiteEventLogger();
var dm = CreateBootedDeploymentManager(siteLog);
await DeployAndAwaitAsync(dm, instanceName, "rev-1", MakeConfigJson(instanceName));
var beforeSweeps = await ResolveInstanceChildAsync(dm, instanceName);
// Rewrite the SAME revision — a same-revision re-store must not read as divergence.
await _storage.StoreDeployedConfigAsync(
instanceName, MakeConfigJson(instanceName), "dep-rev-1", "rev-1", true);
await Task.Delay(SeveralSweeps);
Assert.Equal(beforeSweeps, await ResolveInstanceChildAsync(dm, instanceName));
Assert.Empty(SelfHealEvents(siteLog, instanceName));
}
/// <summary>
/// A stale-revision restart that the site compile gate REJECTS must be announced once and
/// then left alone. Retrying it every tick would be an endless stream of Warning + Error
/// site events for a config that can never apply, and the running instance is unharmed —
/// the gate returns before anything is stopped, so it keeps serving the old config.
/// </summary>
[Fact]
public async Task Sweep_StaleRestartRejectedByTheCompileGate_IsAnnouncedOnceAndNotRetried()
{
const string instanceName = "UncompilablePump";
var siteLog = new FakeSiteEventLogger();
var dm = CreateBootedDeploymentManager(siteLog);
await DeployAndAwaitAsync(dm, instanceName, "rev-1", MakeConfigJson(instanceName));
var beforeHeal = await ResolveInstanceChildAsync(dm, instanceName);
// The refreshed row carries a script that cannot compile.
await _storage.StoreDeployedConfigAsync(
instanceName,
MakeScriptedConfigJson(instanceName, "this is not valid C# at all ((("),
"dep-rev-bad", "rev-bad", true);
// The attempt is announced...
await AwaitAssertAsync(() =>
{
Assert.Single(SelfHealEvents(siteLog, instanceName));
return Task.CompletedTask;
}, TimeSpan.FromSeconds(20));
// ...its failure is announced...
await AwaitAssertAsync(() =>
{
Assert.Contains(siteLog.OfType("deployment"), e =>
e.InstanceId == instanceName &&
e.Severity == "Error" &&
e.Message.Contains("self-heal failed", StringComparison.OrdinalIgnoreCase));
return Task.CompletedTask;
}, TimeSpan.FromSeconds(20));
// ...and several further sweeps add nothing and leave the old actor serving.
await Task.Delay(SeveralSweeps);
Assert.Single(SelfHealEvents(siteLog, instanceName));
Assert.Equal(beforeHeal, await ResolveInstanceChildAsync(dm, instanceName));
Assert.True(await InstanceIsRunningAsync(dm, instanceName));
}
/// <summary>
/// Minimal <see cref="ISiteHealthCollector"/> whose only job is to expose the observable edge
/// "the Deployment Manager's startup config load has completed". The manager calls
/// <see cref="SetInstanceCounts"/> unconditionally the moment that load finishes — including
/// for an empty site — so a non-zero <see cref="ReportCount"/> replaces a wall-clock sleep.
/// </summary>
private sealed class StartupProbe : ISiteHealthCollector
{
private int _reports;
public int ReportCount => Volatile.Read(ref _reports);
public void SetInstanceCounts(int deployed, int enabled, int disabled) =>
Interlocked.Increment(ref _reports);
public void IncrementScriptError() { }
public void IncrementAlarmError() { }
public void IncrementDeadLetter() { }
public void IncrementSiteAuditWriteFailures() { }
public void IncrementAuditRedactionFailure() { }
public void UpdateSiteAuditBacklog(Commons.Types.SiteAuditBacklogSnapshot snapshot) { }
public void UpdateConnectionHealth(string connectionName, ConnectionHealth health) { }
public void RemoveConnection(string connectionName) { }
public void UpdateTagResolution(string connectionName, int totalSubscribed, int successfullyResolved) { }
public void UpdateConnectionEndpoint(string connectionName, string endpoint) { }
public void UpdateTagQuality(string connectionName, int good, int bad, int uncertain) { }
public void SetStoreAndForwardDepths(IReadOnlyDictionary<string, int> depths) { }
public void SetParkedMessageCount(int count) { }
public void SetNodeHostname(string hostname) { }
public void SetClusterNodes(IReadOnlyList<Commons.Messages.Health.NodeStatus> nodes) { }
public void SetActiveNode(bool isActive) { }
public bool IsActiveNode => true;
public Commons.Messages.Health.SiteHealthReport CollectReport(string siteId) =>
throw new NotSupportedException("Startup probe collects no report.");
}
}