fix(deploymentmanager): bulk deploy skips Disabled + reconcile circuit-breaker + probe captures pre-flatten versions (arch-review 2026-08-19 P7-2 + P7-3 + P7-4)

This commit is contained in:
Joseph Doherty
2026-08-19 14:52:00 -04:00
parent b68f553f1e
commit 5cbc143b01
10 changed files with 541 additions and 32 deletions
@@ -45,7 +45,9 @@ public class DeploySiteAsyncTests : TestKit
}
/// <summary>Seeds <paramref name="count"/> deployable instances at the test site.</summary>
private List<Instance> ArrangeInstances(int count)
/// <param name="count">How many instances to seed.</param>
/// <param name="disabledIds">Ids seeded <c>Disabled</c> instead of <c>NotDeployed</c>.</param>
private List<Instance> ArrangeInstances(int count, params int[] disabledIds)
{
var instances = new List<Instance>();
for (var i = 1; i <= count; i++)
@@ -55,7 +57,7 @@ public class DeploySiteAsyncTests : TestKit
Id = i,
SiteId = SiteId,
TemplateId = 10,
State = InstanceState.NotDeployed
State = disabledIds.Contains(i) ? InstanceState.Disabled : InstanceState.NotDeployed
};
instances.Add(instance);
@@ -73,7 +75,11 @@ public class DeploySiteAsyncTests : TestKit
return instances;
}
private DeploymentService CreateService(IActorRef commActor, int maxParallelism, TimeSpan? perInstanceTimeout = null)
private DeploymentService CreateService(
IActorRef commActor,
int maxParallelism,
TimeSpan? perInstanceTimeout = null,
TimeSpan? reconcileTimeout = null)
{
var comms = new CommunicationService(
Options.Create(new CommunicationOptions { DeploymentTimeout = TimeSpan.FromSeconds(30) }),
@@ -84,7 +90,8 @@ public class DeploySiteAsyncTests : TestKit
{
OperationLockTimeout = TimeSpan.FromSeconds(5),
SiteDeploymentMaxParallelism = maxParallelism,
SiteDeploymentTimeoutPerInstance = perInstanceTimeout ?? TimeSpan.FromSeconds(30)
SiteDeploymentTimeoutPerInstance = perInstanceTimeout ?? TimeSpan.FromSeconds(30),
SiteDeploymentReconcileTimeout = reconcileTimeout ?? TimeSpan.FromSeconds(5)
});
return new DeploymentService(
@@ -305,6 +312,147 @@ public class DeploySiteAsyncTests : TestKit
Arg.Any<CancellationToken>());
}
/// <summary>
/// P7-2: a bulk site deploy must not restart deliberately-stopped equipment.
///
/// <para>
/// Deploy is a legal transition out of <c>Disabled</c> and the deploy path
/// forces the instance to <c>Enabled</c> on success, so a site-wide push over an
/// unfiltered instance list silently re-enabled every instance an operator had
/// disabled. Bulk deploy therefore skips them outright — no flatten, no send, no
/// state change — and reports them so the operator can see what was left alone.
/// The single-instance path is unchanged: that is the explicit "yes, re-enable
/// this one" action.
/// </para>
/// </summary>
[Fact]
public async Task DeploySiteAsync_SkipsDisabledInstances_WithoutDeployingOrReEnablingThem()
{
var instances = ArrangeInstances(5, disabledIds: [2, 4]);
var commActor = Sys.ActorOf(Props.Create(() =>
new ThrottledSiteActor(new ConcurrencyTracker(), slowInstanceName: null, slowDelay: TimeSpan.Zero)));
var result = await CreateService(commActor, maxParallelism: 3).DeploySiteAsync(SiteId, "admin");
Assert.True(result.IsSuccess);
var summary = result.Value;
// Only the three Enabled/NotDeployed instances were attempted.
Assert.Equal(3, summary.InstanceResults.Count);
Assert.Equal(3, summary.SuccessCount);
Assert.Equal(0, summary.FailureCount);
Assert.DoesNotContain(summary.InstanceResults, r => r.UniqueName is "Inst-02" or "Inst-04");
// The Disabled pair is reported as skipped, with the reason the operator needs.
Assert.NotNull(summary.SkippedInstances);
var skipped = summary.SkippedInstances;
Assert.Equal(2, skipped.Count);
Assert.Equal(
new[] { "Inst-02", "Inst-04" },
skipped.Select(s => s.UniqueName).Order().ToArray());
Assert.All(skipped, s => Assert.Contains("deploy individually to re-enable", s.Reason));
// ...and left completely untouched: still Disabled, and never handed to the
// repository at all (no deployment record, no instance update, no snapshot).
foreach (var id in new[] { 2, 4 })
{
Assert.Equal(InstanceState.Disabled, instances.Single(i => i.Id == id).State);
await _pipeline.DidNotReceive().FlattenAndValidateAsync(
id, Arg.Any<CancellationToken>(), Arg.Any<bool>(), Arg.Any<FlattenSession?>());
}
await _repo.DidNotReceive().UpdateInstanceAsync(
Arg.Is<Instance>(i => i.Id == 2 || i.Id == 4), Arg.Any<CancellationToken>());
await _repo.DidNotReceive().AddDeploymentRecordAsync(
Arg.Is<DeploymentRecord>(r => r.InstanceId == 2 || r.InstanceId == 4), Arg.Any<CancellationToken>());
}
/// <summary>
/// P7-3: phase 1 is serial, so an unreachable site used to cost the batch one
/// full 30-second <c>QueryTimeout</c> PER INSTANCE before a single deploy was
/// sent — hours for a large site, so the CLI's 5-minute bulk deadline cancelled
/// and the next attempt re-armed the same cost. Two bounds close it: a short
/// per-reconcile deadline, and a once-per-batch circuit-breaker that skips the
/// remaining reconciles after the first timeout.
///
/// <para>
/// The site here answers the deploy but NEVER answers the reconcile query, so
/// the assertion that exactly ONE query was issued for six instances is the
/// whole finding: without the breaker there would be six.
/// </para>
/// </summary>
[Fact]
public async Task DeploySiteAsync_UnreachableSite_ReconcilesOnceThenSkipsTheRestOfTheBatch()
{
const int instanceCount = 6;
ArrangeInstances(instanceCount);
// Reconciliation only runs when a prior record is stuck InProgress.
for (var i = 1; i <= instanceCount; i++)
{
_repo.GetCurrentDeploymentStatusAsync(i, Arg.Any<CancellationToken>())
.Returns(new DeploymentRecord($"prior-{i}", "admin")
{
InstanceId = i,
Status = DeploymentStatus.InProgress,
RevisionHash = "sha256:old"
});
}
var queryCount = 0;
var commActor = Sys.ActorOf(Props.Create(() =>
new SilentQuerySiteActor(() => Interlocked.Increment(ref queryCount))));
var reconcileTimeout = TimeSpan.FromMilliseconds(300);
var service = CreateService(commActor, maxParallelism: 3, reconcileTimeout: reconcileTimeout);
var sw = Stopwatch.StartNew();
var result = await service.DeploySiteAsync(SiteId, "admin");
sw.Stop();
Assert.True(result.IsSuccess);
// The circuit-breaker opened after the first unanswered reconcile.
Assert.Equal(1, Volatile.Read(ref queryCount));
// Prepare therefore cost ONE reconcile deadline, not six.
Assert.True(sw.Elapsed < TimeSpan.FromSeconds(2),
$"bulk prepare took {sw.ElapsedMilliseconds} ms — the per-batch reconcile bound is not holding");
// A skipped reconcile is not a failed deploy: every instance still went out.
Assert.Equal(instanceCount, result.Value.SuccessCount);
Assert.Equal(0, result.Value.FailureCount);
}
/// <summary>
/// Answers <c>RefreshDeploymentCommand</c> normally but NEVER answers a
/// <c>DeploymentStateQueryRequest</c>, counting how many arrived — a site that
/// is up for deploys yet unreachable for the reconcile query, which is the shape
/// a wedged or overloaded site presents.
/// </summary>
private sealed class SilentQuerySiteActor : ReceiveActor
{
public SilentQuerySiteActor(Action onQuery)
{
Receive<SiteEnvelope>(env =>
{
switch (env.Message)
{
case DeploymentStateQueryRequest:
onQuery();
return;
case RefreshDeploymentCommand cmd:
Sender.Tell(new DeploymentStatusResponse(
cmd.DeploymentId, cmd.InstanceUniqueName,
DeploymentStatus.Success, null, DateTimeOffset.UtcNow));
return;
}
});
}
}
/// <summary>
/// Answers every <c>RefreshDeploymentCommand</c> after a fixed delay, recording
/// how old each command's staging <c>Timestamp</c> already was on arrival.
@@ -53,4 +53,23 @@ public class DeploymentManagerOptionsValidatorTests
Assert.True(result.Failed);
Assert.Contains("OperationLockTimeout", result.FailureMessage);
}
[Fact]
public void ZeroSiteDeploymentReconcileTimeout_IsRejected()
{
var result = Validate(new DeploymentManagerOptions { SiteDeploymentReconcileTimeout = TimeSpan.Zero });
Assert.True(result.Failed);
Assert.Contains("SiteDeploymentReconcileTimeout", result.FailureMessage);
}
[Fact]
public void NegativeSiteDeploymentReconcileTimeout_IsRejected()
{
var result = Validate(new DeploymentManagerOptions
{
SiteDeploymentReconcileTimeout = TimeSpan.FromSeconds(-1)
});
Assert.True(result.Failed);
Assert.Contains("SiteDeploymentReconcileTimeout", result.FailureMessage);
}
}
@@ -1597,6 +1597,36 @@ public class DeploymentServiceTests : TestKit
Assert.Equal(InstanceState.Disabled, instance.State);
}
/// <summary>
/// The counterpart to the bulk path's Disabled skip (arch-review P7-2): a
/// SINGLE-instance deploy of a Disabled instance still deploys AND enables it.
/// That is the documented "Deploy on a Disabled instance also enables it"
/// transition and the explicit operator action bulk deploy now refuses to
/// perform implicitly, so it must keep working exactly as before.
/// </summary>
[Fact]
public async Task DeployInstanceAsync_DisabledInstance_StillDeploysAndEnables()
{
var instance = new Instance("SingleDisabled")
{
Id = 73, SiteId = 1, State = InstanceState.Disabled
};
_repo.GetInstanceByIdAsync(73, Arg.Any<CancellationToken>()).Returns(instance);
SetupValidPipeline(73, "SingleDisabled", "sha256:target");
_repo.GetCurrentDeploymentStatusAsync(73, Arg.Any<CancellationToken>())
.Returns((DeploymentRecord?)null);
var counters = new ReconcileProbeCounters();
var commActor = Sys.ActorOf(Props.Create(() =>
new ReconcileProbeActor(counters, siteHash: "sha256:target", failQuery: false)));
var result = await CreateServiceWithCommActor(commActor).DeployInstanceAsync(73, "admin");
Assert.True(result.IsSuccess);
Assert.Equal(1, counters.DeployCount);
Assert.Equal(InstanceState.Enabled, instance.State);
}
// ── DeploymentManager-016: reconciled record must carry the target revision hash ──
[Fact]
@@ -1,4 +1,5 @@
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Templates;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Templates;
@@ -96,4 +97,72 @@ public class StaleInstanceProbeTests
Assert.Equal("sha256:second", await probe.GetCurrentRevisionHashAsync(InstanceId));
}
/// <summary>
/// P7-4: a non-structural template edit landing DURING the flatten must not be
/// captured as if it were already included.
///
/// <para>
/// The memo pairs a revision hash with the watermark readings that produced it.
/// Reading the per-template versions AFTER the flatten pairs the PRE-edit hash
/// with the POST-edit version, and every later probe then finds the watermark
/// unchanged and answers "not stale" forever — until an unrelated edit, a
/// restart, or the 4096-entry overflow clears it. The staleness sweep and
/// <c>ImportResult.StaleInstanceIds</c> both read through this memo.
/// </para>
///
/// <para>
/// The fix is that the versions come from the session, which records each
/// template's version at the instant it READ that template — before the edit —
/// so the mismatch on the next probe invalidates the memo. Only the recording is
/// changed: the mid-flatten edit still produces the pre-edit hash on this pass,
/// which is correct, since that is the graph the flatten actually saw.
/// </para>
/// </summary>
[Fact]
public async Task TemplateEditedDuringFlatten_InvalidatesTheMemo_RatherThanPinningTheStaleHash()
{
const int TemplateId = 42;
// The flatten reads the template through the session (recording the version
// it read it at), and the edit lands before the flatten returns.
_pipeline
.FlattenAndValidateAsync(InstanceId, Arg.Any<CancellationToken>(), Arg.Any<bool>(), Arg.Any<FlattenSession?>())
.Returns(async call =>
{
_flattenCount++;
var session = call.ArgAt<FlattenSession?>(3)!;
await session.GetTemplateAsync(TemplateId, _ => Task.FromResult<Template?>(new Template($"T{TemplateId}")));
// A non-structural template edit commits while the flatten is in flight.
_watermark.BumpTemplate(TemplateId);
_currentHash = "sha256:after-edit";
var config = new FlattenedConfiguration { InstanceUniqueName = "Inst-01" };
return Result<FlatteningPipelineResult>.Success(
new FlatteningPipelineResult(config, "sha256:before-edit", ValidationResult.Success()));
});
var probe = CreateProbe();
// Pass 1 legitimately returns the hash of the graph the flatten walked.
Assert.Equal("sha256:before-edit", await probe.GetCurrentRevisionHashAsync(InstanceId));
Assert.Equal(1, _flattenCount);
// Pass 2 must see the edit. Re-recording the versions after the flatten
// would have memoised the post-edit version against the pre-edit hash, and
// this call would still answer "sha256:before-edit" without re-flattening.
_pipeline
.FlattenAndValidateAsync(InstanceId, Arg.Any<CancellationToken>(), Arg.Any<bool>(), Arg.Any<FlattenSession?>())
.Returns(_ =>
{
_flattenCount++;
var config = new FlattenedConfiguration { InstanceUniqueName = "Inst-01" };
return Result<FlatteningPipelineResult>.Success(
new FlatteningPipelineResult(config, _currentHash, ValidationResult.Success()));
});
Assert.Equal("sha256:after-edit", await probe.GetCurrentRevisionHashAsync(InstanceId));
Assert.Equal(2, _flattenCount);
}
}