fix(siteruntime): extend the row-101 stale-snapshot guard to the instance-config half
ReadStoreSnapshotAsync's snapshot is read off the actor thread and consumed only when its piped message lands. Row 101 guarded the DEFINITIONS half of that race; the CONFIG half was left to machinery whose reach is time-limited, and both legs of the instance reconcile sweep were reachable through it: - Row-72 leg: a read dispatched before a deploy's persist, landing after the deploy completed AND after the pipeline released ownership, carries the revision the deploy replaced. TryRestartStaleInstance then "repairs" the instance BACKWARD, re-deploying the snapshot's own superseded ConfigJson. InstanceIsOwnedByADeployPipeline covers only in-flight deploys, and the per-revision claim has never seen the old revision, so nothing stopped it. - Row-78 leg: a read dispatched before a delete/disable commits still shows an enabled row while the actor is already stopped, so StartStoredInstance resurrects it — and delete clears _sweepAttemptedRevisions, freeing the claim. Both verified real by neutralising the freshness check and re-running the new tests (the fail-before site event literally reads "running revision rev-2 is stale, applying stored revision rev-1"; the deleted instance came back). _definitionsWriteGeneration becomes _storeWriteGeneration, covering both tables (the snapshot is one consistent read and is discarded whole, so one counter is the right shape); DefinitionsGenerationAtRead -> StoreGenerationAtRead on both internal records. The generation is now bumped at BOTH ends of every mutation — when the off-thread write is dispatched and when its completion message lands — because the config paths advance actor state before the write, so an initiation-only bump would let an in-window read look fresh. Disable, delete and the init-failure rollback have no actor-thread completion message of their own, so a minimal StoreWriteCompleted self-message carries the bump for them. CDC-replicated writes, snapshot resyncs, restored files and SiteReconciliationActor's fetch deliberately do NOT bump: discovering exactly those is the sweep's reason to exist. A stale sweep result is now dropped WHOLE (no overlay apply, no start, no restart). A stale boot load cannot be dropped — startup must finish — so it re-dispatches LoadDeployedConfigs(); a fifth consecutive superseded read is applied anyway with a Warning, and a late one arriving after the boot completed is dropped rather than looping against the S5 retry timer. Tests: three fail-before regressions (backward restart, resurrection, boot re-read) plus two guard twins pinning that the legitimate row-72 restart and row-78 start still fire on a current-generation snapshot. The row-101 definitions tests keep passing; FreshEmptySweepSnapshot's hardcoded generation moves 1 -> 2 now that one artifact apply bumps at both ends. Docs: Component-SiteRuntime.md's stale-snapshot paragraph now describes the whole snapshot (both halves, both consumers, the boot re-dispatch, the CDC-no-bump rule), the components mirror follows, and the arbitration memo gains §11 extending §10 to the config half.
This commit is contained in:
Joseph Doherty
+6
-3
@@ -596,7 +596,7 @@ public class DclConnectionArbitrationTests : TestKit, IDisposable
|
||||
|
||||
// Generation 0 = captured before the apply bumped it; empty list = the pre-write table.
|
||||
dm.Tell(new DeploymentManagerActor.InstanceSweepLoaded(
|
||||
[], null, ConnectionDefinitions: [], DefinitionsGenerationAtRead: 0));
|
||||
[], null, ConnectionDefinitions: [], StoreGenerationAtRead: 0));
|
||||
|
||||
await Task.Delay(TimeSpan.FromSeconds(2));
|
||||
Assert.DoesNotContain(_removes, r => r.ConnectionName == ConnectionName);
|
||||
@@ -619,9 +619,12 @@ public class DclConnectionArbitrationTests : TestKit, IDisposable
|
||||
Assert.True(ExpectMsg<ArtifactDeploymentResponse>(TimeSpan.FromSeconds(20)).Success);
|
||||
await AwaitCreateCountAsync(1, "the artifact apply never created the connection");
|
||||
|
||||
// Exactly one artifact apply has run, so the current write generation is 1.
|
||||
// Exactly one artifact apply has run, and every store mutation bumps the generation
|
||||
// TWICE — once when the off-thread write is dispatched and once when its completion
|
||||
// message lands (that pair is what leaves no window in which an in-flight read looks
|
||||
// fresh; see DeploymentManagerActor._storeWriteGeneration) — so the current value is 2.
|
||||
dm.Tell(new DeploymentManagerActor.InstanceSweepLoaded(
|
||||
[], null, ConnectionDefinitions: [], DefinitionsGenerationAtRead: 1));
|
||||
[], null, ConnectionDefinitions: [], StoreGenerationAtRead: 2));
|
||||
|
||||
await AwaitAssertAsync(() =>
|
||||
{
|
||||
|
||||
@@ -4,6 +4,7 @@ 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.Messages.Lifecycle;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
|
||||
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
|
||||
@@ -510,4 +511,257 @@ public class InstanceSweepSelfHealTests : TestKit, IDisposable
|
||||
Assert.True(await InstanceIsRunningAsync(dm, instanceName));
|
||||
}
|
||||
|
||||
// ── Row 101 (config half): a sweep snapshot older than the actor's own state ──
|
||||
|
||||
/// <summary>
|
||||
/// A crafted sweep result standing in for a read that was dispatched earlier than the
|
||||
/// generation it is stamped with. <paramref name="generationAtRead"/> <c>0</c> is the value a
|
||||
/// read dispatched before this actor had mutated anything would carry; anything the actor has
|
||||
/// done since makes it stale. Connection definitions are passed as <see langword="null"/>
|
||||
/// (= "the definitions read failed, keep the overlay") so these tests exercise the CONFIG
|
||||
/// half of the snapshot in isolation.
|
||||
/// </summary>
|
||||
private static DeploymentManagerActor.InstanceSweepLoaded SweepSnapshot(
|
||||
string instanceName, string revision, string configJson, long generationAtRead,
|
||||
bool isEnabled = true) =>
|
||||
new(
|
||||
[
|
||||
new DeployedInstance
|
||||
{
|
||||
InstanceUniqueName = instanceName,
|
||||
ConfigJson = configJson,
|
||||
DeploymentId = $"dep-{revision}",
|
||||
RevisionHash = revision,
|
||||
IsEnabled = isEnabled,
|
||||
DeployedAt = DateTimeOffset.UtcNow.ToString("O")
|
||||
}
|
||||
],
|
||||
null,
|
||||
ConnectionDefinitions: null,
|
||||
StoreGenerationAtRead: generationAtRead);
|
||||
|
||||
/// <summary>
|
||||
/// <b>Deferred register row 101, the row-72 leg.</b> A sweep read dispatched before a deploy
|
||||
/// persisted its new row can land after that deploy has completed AND after the deploy
|
||||
/// pipeline released the instance — <see cref="DeploymentManagerActor"/>'s
|
||||
/// <c>InstanceIsOwnedByADeployPipeline</c> only covers a deploy still in flight, so it is no
|
||||
/// defence here. The snapshot then carries the revision the deploy just REPLACED while the
|
||||
/// running actor carries the new one, which is bit-for-bit the divergence the row-72 repair
|
||||
/// looks for — in the wrong direction.
|
||||
///
|
||||
/// <para>
|
||||
/// <b>Fails before the fix</b>: the sweep "repairs" the instance BACKWARD, re-deploying the
|
||||
/// snapshot's own superseded config JSON through the redeploy path (attribute back to
|
||||
/// <c>111</c>) and announcing it as a self-heal. Nothing retries it either — the per-revision
|
||||
/// claim marks the OLD revision as attempted, so the instance simply stays reverted.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task StaleSweepSnapshot_CarryingASupersededRevision_DoesNotRestartBackward()
|
||||
{
|
||||
const string instanceName = "BackwardRestartPump";
|
||||
var siteLog = new FakeSiteEventLogger();
|
||||
// Sweep effectively disabled: the crafted message must be the only sweep result in play.
|
||||
var dm = CreateBootedDeploymentManager(siteLog, sweepSeconds: 3600);
|
||||
|
||||
await DeployAndAwaitAsync(
|
||||
dm, instanceName, "rev-1", MakeConfigJson(instanceName, attributeValue: "111"));
|
||||
await DeployAndAwaitAsync(
|
||||
dm, instanceName, "rev-2", MakeConfigJson(instanceName, attributeValue: "222"));
|
||||
|
||||
var afterDeploys = await ResolveInstanceChildAsync(dm, instanceName);
|
||||
|
||||
// Generation 0 = captured before either deploy mutated the store.
|
||||
dm.Tell(SweepSnapshot(
|
||||
instanceName, "rev-1", MakeConfigJson(instanceName, attributeValue: "111"),
|
||||
generationAtRead: 0));
|
||||
|
||||
await Task.Delay(SeveralSweeps);
|
||||
|
||||
Assert.Empty(SelfHealEvents(siteLog, instanceName));
|
||||
Assert.Equal(afterDeploys, await ResolveInstanceChildAsync(dm, instanceName));
|
||||
var snapshot = await SnapshotAsync(dm, instanceName);
|
||||
Assert.Contains(snapshot.AttributeValues, a =>
|
||||
a.AttributeName == "TestAttr" && a.Value?.ToString() == "222");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The row-101 guard must not blunt the row-72 repair it sits in front of: the SAME message
|
||||
/// stamped with the CURRENT generation is a genuine answer and must still converge the
|
||||
/// runtime on the stored revision.
|
||||
///
|
||||
/// <para>
|
||||
/// The instance is started by the BOOT LOAD from a pre-written row rather than by a deploy
|
||||
/// command, precisely so no store mutation passes through the actor and the current
|
||||
/// generation is still <c>0</c> — the same literal the stale test uses, with the opposite
|
||||
/// meaning, which is the whole point of the guard.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task FreshSweepSnapshot_CarryingANewerRevision_StillRestartsTheInstance()
|
||||
{
|
||||
const string instanceName = "FreshRestartPump";
|
||||
var siteLog = new FakeSiteEventLogger();
|
||||
|
||||
await _storage.StoreDeployedConfigAsync(
|
||||
instanceName, MakeConfigJson(instanceName, attributeValue: "111"),
|
||||
"dep-rev-1", "rev-1", true);
|
||||
|
||||
var dm = CreateBootedDeploymentManager(siteLog, sweepSeconds: 3600);
|
||||
await AwaitAssertAsync(async () =>
|
||||
Assert.True(await InstanceIsRunningAsync(dm, instanceName)), TimeSpan.FromSeconds(20));
|
||||
|
||||
dm.Tell(SweepSnapshot(
|
||||
instanceName, "rev-2", MakeConfigJson(instanceName, attributeValue: "222"),
|
||||
generationAtRead: 0));
|
||||
|
||||
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));
|
||||
|
||||
Assert.Contains(SelfHealEvents(siteLog, instanceName), e =>
|
||||
e.Severity == "Warning" &&
|
||||
e.Message.Contains("rev-1", StringComparison.Ordinal) &&
|
||||
e.Message.Contains("rev-2", StringComparison.Ordinal));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <b>Deferred register row 101, the row-78 leg.</b> A sweep read dispatched before a delete
|
||||
/// committed still shows the instance's row — enabled, with a config — while the actor has
|
||||
/// already stopped it and dropped its bookkeeping. The delete specifically clears
|
||||
/// <c>_sweepAttemptedRevisions</c>, so the once-per-revision claim is freshly available and
|
||||
/// does not stand in the way either.
|
||||
///
|
||||
/// <para>
|
||||
/// <b>Fails before the fix</b>: the sweep RESURRECTS the instance the operator just deleted,
|
||||
/// re-adds it to the deployed-name set, and announces the resurrection as a self-heal.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task StaleSweepSnapshot_ShowingADeletedInstance_DoesNotResurrectIt()
|
||||
{
|
||||
const string instanceName = "ResurrectedPump";
|
||||
var siteLog = new FakeSiteEventLogger();
|
||||
var dm = CreateBootedDeploymentManager(siteLog, sweepSeconds: 3600);
|
||||
|
||||
await DeployAndAwaitAsync(dm, instanceName, "rev-1", MakeConfigJson(instanceName));
|
||||
|
||||
dm.Tell(new DeleteInstanceCommand("cmd-del-1", instanceName, DateTimeOffset.UtcNow));
|
||||
Assert.True(ExpectMsg<InstanceLifecycleResponse>(TimeSpan.FromSeconds(15)).Success);
|
||||
await AwaitAssertAsync(async () =>
|
||||
Assert.False(await InstanceIsRunningAsync(dm, instanceName)), TimeSpan.FromSeconds(15));
|
||||
|
||||
// Generation 0 = captured before the deploy and the delete mutated the store.
|
||||
dm.Tell(SweepSnapshot(
|
||||
instanceName, "rev-1", MakeConfigJson(instanceName), generationAtRead: 0));
|
||||
|
||||
await Task.Delay(SeveralSweeps);
|
||||
|
||||
Assert.False(await InstanceIsRunningAsync(dm, instanceName));
|
||||
Assert.Empty(SelfHealEvents(siteLog, instanceName));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The mirror of the resurrection test: the same enabled-config-with-no-actor snapshot,
|
||||
/// stamped with the CURRENT generation, must still perform the row-78 start. Nothing has
|
||||
/// mutated the store through this actor, so <c>0</c> is current here.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task FreshSweepSnapshot_ShowingAStoredEnabledInstance_StillStartsIt()
|
||||
{
|
||||
const string instanceName = "FreshStartPump";
|
||||
var siteLog = new FakeSiteEventLogger();
|
||||
var dm = CreateBootedDeploymentManager(siteLog, sweepSeconds: 3600);
|
||||
|
||||
dm.Tell(SweepSnapshot(
|
||||
instanceName, "rev-1", MakeConfigJson(instanceName), generationAtRead: 0));
|
||||
|
||||
await AwaitAssertAsync(async () =>
|
||||
Assert.True(await InstanceIsRunningAsync(dm, instanceName)), TimeSpan.FromSeconds(20));
|
||||
Assert.Single(SelfHealEvents(siteLog, instanceName));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <b>Row 101, boot half.</b> The boot-time load has to COMPLETE — a node whose startup read
|
||||
/// is simply dropped stays a silent zero-instance node — so a superseded boot snapshot is
|
||||
/// re-read rather than discarded like the sweep's.
|
||||
///
|
||||
/// <para>
|
||||
/// The race is made deterministic by a slow first read: a lifecycle command that mutates
|
||||
/// <c>deployed_configurations</c> is processed while it is in flight, so the snapshot lands
|
||||
/// stamped with a generation the actor has already moved past. The observable is the read
|
||||
/// itself — the loader must be called a SECOND time — and, just as importantly, exactly twice:
|
||||
/// the re-dispatch captures the post-mutation generation, so it converges instead of looping.
|
||||
/// </para>
|
||||
///
|
||||
/// <para><b>Fails before the fix</b>: one read, and the pre-mutation snapshot is applied.</para>
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task StaleBootSnapshot_IsReReadRatherThanApplied()
|
||||
{
|
||||
var reads = 0;
|
||||
var health = new DeploymentManagerStartupProbe();
|
||||
|
||||
var configLoader = SlowFirstConfigLoad(() => Interlocked.Increment(ref reads));
|
||||
var dm = ActorOf(Props.Create(() => new DeploymentManagerActor(
|
||||
_storage,
|
||||
_compilationService,
|
||||
_sharedScriptLibrary,
|
||||
null,
|
||||
new SiteRuntimeOptions { InstanceSweepIntervalSeconds = 3600 },
|
||||
NullLogger<DeploymentManagerActor>.Instance,
|
||||
dclManager: null,
|
||||
healthCollector: health,
|
||||
serviceProvider: null,
|
||||
loggerFactory: null,
|
||||
configFetcher: null,
|
||||
startupLoadRetryInterval: null,
|
||||
configLoader: configLoader,
|
||||
externalSystemCache: null,
|
||||
scriptScheduler: null,
|
||||
connectionDefinitionLoader: null)));
|
||||
|
||||
// Mutates deployed_configurations (a no-op UPDATE for an unknown instance is still a
|
||||
// write dispatched from the actor thread) while the first read is still in flight.
|
||||
dm.Tell(new DisableInstanceCommand("cmd-boot-race", "NoSuchInstance", DateTimeOffset.UtcNow));
|
||||
ExpectMsg<InstanceLifecycleResponse>(TimeSpan.FromSeconds(15));
|
||||
|
||||
await AwaitAssertAsync(() =>
|
||||
{
|
||||
Assert.Equal(2, Volatile.Read(ref reads));
|
||||
return Task.CompletedTask;
|
||||
}, TimeSpan.FromSeconds(20));
|
||||
|
||||
// The second read is fresh, so the boot completes and stops re-reading...
|
||||
await Task.Delay(TimeSpan.FromSeconds(2));
|
||||
Assert.Equal(2, Volatile.Read(ref reads));
|
||||
|
||||
// ...and a LATE superseded snapshot (the S5 retry timer racing a success) is dropped
|
||||
// rather than triggering another read, because the boot it would complete is done.
|
||||
dm.Tell(new DeploymentManagerActor.StartupConfigsLoaded(
|
||||
[], null, ConnectionDefinitions: null, StoreGenerationAtRead: -1));
|
||||
await Task.Delay(TimeSpan.FromSeconds(2));
|
||||
Assert.Equal(2, Volatile.Read(ref reads));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A deployed-config loader whose FIRST call takes long enough for a lifecycle command to be
|
||||
/// processed while it is in flight; later calls answer immediately. Always returns an empty
|
||||
/// site — these tests assert on how many reads happen, not on what they carry.
|
||||
/// </summary>
|
||||
private static Func<Task<List<DeployedInstance>>> SlowFirstConfigLoad(Action onRead)
|
||||
{
|
||||
var calls = 0;
|
||||
return async () =>
|
||||
{
|
||||
onRead();
|
||||
if (Interlocked.Increment(ref calls) == 1)
|
||||
await Task.Delay(TimeSpan.FromSeconds(2));
|
||||
return [];
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user