fix(site-runtime): serialise DeployArtifactsCommand applies per site so an overlapping apply cannot delete a live artifact (review 2026-10-21 01/SR-C2)
Each apply upserts its full set then deletes everything else, off the actor thread; two overlapping commands interleaved and deleted artifacts central never removed. The Deployment Manager now runs one apply at a time, queuing the rest in arrival order; the slot is released by a self-message that lands behind the apply's own DCL push.
This commit is contained in:
@@ -165,6 +165,7 @@ flowchart TD
|
||||
- Stores all artifacts in local SQLite. After artifact deployment, the site is fully self-contained — all runtime configuration is read from local SQLite with no access to the central configuration database.
|
||||
- Recompiles shared scripts and makes updated code available to all Script Actors.
|
||||
- **Reconciles deletions**: each non-null artifact list is the complete set for its class, so after upserting, any stored row absent from the set is deleted (the artifact was deleted centrally — there is no per-artifact delete command). A removed shared script is also unregistered from the compiled script library, and a removed data connection's live DCL connection actor is stopped. A `null` list means "field not shipped" and reconciles nothing.
|
||||
- **Applies are serialised per site, in arrival order (review 2026-10-21 01/SR-C2).** Because each apply is "upsert my complete set, then delete every stored row not in it" and runs off the actor thread, two overlapping `DeployArtifactsCommand`s (central's per-site retry racing a fresh deploy, two operators clicking *Deploy Artifacts*) used to run concurrently, and the interleaving *B upserts Z → A deletes-except(A's set)* deleted Z — a shared script, external system, data connection or deployed secret central never removed — until the next artifact deploy. The Deployment Manager now keeps one apply in flight and queues the rest; the slot is released by a self-message sent after the apply task completes, i.e. behind that apply's own DCL push, so the final state is always exactly the last-arriving command's set. No revision comparison is made: a genuinely *older* snapshot that arrives last still wins, which is central's ordering to guarantee.
|
||||
|
||||
### Deployed Secrets (2026-09-28)
|
||||
- `HandleDeployArtifacts` additionally accepts the command's `Secrets` list (`SecretArtifact`, ciphertext only) and writes the full shipped set to the `deployed_secrets` table via `SiteStorageService.UpsertDeployedSecretsAsync` — upsert every shipped name, soft-remove (`is_removed = 1`) any stored name absent from the set. A `null` `Secrets` field (pre-feature central) touches nothing and answers with no `SecretsApplied`. A current central always ships a list — **empty** when the site references no secrets — so dropping a site's last `secret:` reference removes the secret from both nodes. Storage detail: *Deployed Secrets (Site SQLite / LocalDb)* below.
|
||||
|
||||
@@ -419,6 +419,8 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
|
||||
|
||||
// Handle system-wide artifact deployment
|
||||
Receive<DeployArtifactsCommand>(HandleDeployArtifacts);
|
||||
// Review 2026-10-21 01/SR-C2: releases the single artifact-apply slot.
|
||||
Receive<ArtifactApplyCompleted>(_ => HandleArtifactApplyCompleted());
|
||||
|
||||
// Artifact-deploy DCL push, dispatched back from the
|
||||
// off-thread persistence task so the hash-cache mutation stays
|
||||
@@ -3267,16 +3269,62 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
|
||||
}).PipeTo(sender);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Artifact applies waiting for the one in flight to finish (review 2026-10-21 01/SR-C2), in
|
||||
/// arrival order, each with the sender to answer. See <see cref="HandleDeployArtifacts"/>.
|
||||
/// </summary>
|
||||
private readonly Queue<(DeployArtifactsCommand Command, IActorRef ReplyTo)> _queuedArtifactApplies = new();
|
||||
|
||||
/// <summary>True while one artifact apply's off-thread task is running.</summary>
|
||||
private bool _artifactApplyInFlight;
|
||||
|
||||
/// <summary>
|
||||
/// Handles system-wide artifact deployment (shared scripts, external systems, etc.).
|
||||
/// Persists artifacts to SiteStorageService and recompiles shared scripts.
|
||||
///
|
||||
/// <para>
|
||||
/// <b>Applies are serialised, one at a time, in arrival order (review 2026-10-21
|
||||
/// 01/SR-C2).</b> Each apply is "upsert my complete set, then delete every stored row not in
|
||||
/// my set", and it runs off the actor thread. Two commands that overlapped — central's per-site
|
||||
/// retry racing a fresh deploy, or two operators clicking <i>Deploy Artifacts</i> — used to run
|
||||
/// CONCURRENTLY, so the interleaving "B upserts Z → A deletes-except(A's set)" deleted Z (and
|
||||
/// unregistered it from the shared-script library, or soft-removed a deployed secret) although
|
||||
/// central never removed it, and the site stayed wrong until the next artifact deploy. Queued
|
||||
/// here, the final state is always exactly one command's set: the last one to arrive.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
private void HandleDeployArtifacts(DeployArtifactsCommand command)
|
||||
{
|
||||
if (_artifactApplyInFlight)
|
||||
{
|
||||
_queuedArtifactApplies.Enqueue((command, Sender));
|
||||
_logger.LogInformation(
|
||||
"Artifact deployment {DeploymentId} queued behind the apply in flight ({Queued} waiting)",
|
||||
command.DeploymentId, _queuedArtifactApplies.Count);
|
||||
return;
|
||||
}
|
||||
|
||||
StartArtifactApply(command, Sender);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The in-flight artifact apply has finished (its reply and any DCL push are already on
|
||||
/// their way); start the next queued one, if any.
|
||||
/// </summary>
|
||||
private void HandleArtifactApplyCompleted()
|
||||
{
|
||||
_artifactApplyInFlight = false;
|
||||
if (_queuedArtifactApplies.TryDequeue(out var next))
|
||||
StartArtifactApply(next.Command, next.ReplyTo);
|
||||
}
|
||||
|
||||
/// <summary>Runs one artifact apply off the actor thread; see <see cref="HandleDeployArtifacts"/>.</summary>
|
||||
private void StartArtifactApply(DeployArtifactsCommand command, IActorRef sender)
|
||||
{
|
||||
_artifactApplyInFlight = true;
|
||||
_logger.LogInformation(
|
||||
"Deploying system artifacts, deploymentId={DeploymentId}", command.DeploymentId);
|
||||
|
||||
var sender = Sender;
|
||||
// Capture Self before entering Task.Run: the Self/Sender/Context properties
|
||||
// are backed by the ambient ActorCell, which is null on a thread-pool thread,
|
||||
// so reading Self *inside* the lambda throws "no active ActorContext". The
|
||||
@@ -3289,7 +3337,7 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
|
||||
// pre-apply table and be treated as current.
|
||||
BumpStoreWriteGeneration("artifact apply dispatch");
|
||||
|
||||
Task.Run(async () =>
|
||||
var apply = Task.Run(async () =>
|
||||
{
|
||||
// The DCL push for this deploy's data connections is captured
|
||||
// here and sent only AFTER the shipped secrets are written and mirrored (in the
|
||||
@@ -3508,7 +3556,13 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
|
||||
self.Tell(pendingDclApply);
|
||||
}
|
||||
}
|
||||
}).PipeTo(sender);
|
||||
});
|
||||
apply.PipeTo(sender);
|
||||
|
||||
// Releases the single apply slot. Told after the task completes, so it is enqueued behind
|
||||
// this apply's own ApplyArtifactDataConnectionsToDcl (sent from the finally above) — the
|
||||
// next apply never starts before this one's DCL push has been processed.
|
||||
apply.ContinueWith(_ => ArtifactApplyCompleted.Instance).PipeTo(self);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -3916,6 +3970,17 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
|
||||
/// carries the reconcile-deleted definitions so the hash-cache eviction and
|
||||
/// live DCL connection stop also happen actor-thread-confined.
|
||||
/// </summary>
|
||||
/// <summary>
|
||||
/// Review 2026-10-21 01/SR-C2: piped to self when an artifact apply's task has completed, so
|
||||
/// the next queued <see cref="DeployArtifactsCommand"/> may start.
|
||||
/// </summary>
|
||||
internal sealed record ArtifactApplyCompleted
|
||||
{
|
||||
/// <summary>The shared singleton instance.</summary>
|
||||
public static readonly ArtifactApplyCompleted Instance = new();
|
||||
private ArtifactApplyCompleted() { }
|
||||
}
|
||||
|
||||
internal record ApplyArtifactDataConnectionsToDcl(
|
||||
IReadOnlyList<Commons.Messages.Artifacts.DataConnectionArtifact> DataConnections,
|
||||
IReadOnlyList<string> RemovedConnectionNames);
|
||||
|
||||
+101
@@ -0,0 +1,101 @@
|
||||
using Akka.Actor;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Artifacts;
|
||||
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>
|
||||
/// Review 2026-10-21 01/SR-C2: each <see cref="DeployArtifactsCommand"/> apply is "upsert my
|
||||
/// full set, then delete everything not in my set", and two overlapping commands used to run
|
||||
/// CONCURRENTLY on the thread pool. The interleaving "B upserts Z → A deletes-except(A's set)"
|
||||
/// left a centrally-live artifact deleted on the site. Applies are now serialised per site, in
|
||||
/// arrival order.
|
||||
/// </summary>
|
||||
public class DeploymentManagerArtifactSerializationTests : TestKit, IDisposable
|
||||
{
|
||||
private readonly SiteStorageService _storage;
|
||||
private readonly ScriptCompilationService _compilationService;
|
||||
private readonly SharedScriptLibrary _sharedScriptLibrary;
|
||||
private readonly TestLocalDb _localDb;
|
||||
|
||||
/// <summary>Creates a temporary LocalDb-backed site store per test.</summary>
|
||||
public DeploymentManagerArtifactSerializationTests()
|
||||
{
|
||||
_localDb = TestLocalDb.CreateTemp("dm-artifact-serial");
|
||||
_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);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
void IDisposable.Dispose()
|
||||
{
|
||||
Shutdown();
|
||||
var path = _localDb.Path;
|
||||
_localDb.Dispose();
|
||||
TestLocalDb.DeleteFiles(path);
|
||||
}
|
||||
|
||||
private IActorRef CreateBootedDeploymentManager()
|
||||
{
|
||||
var health = new DeploymentManagerStartupProbe();
|
||||
var dm = ActorOf(Props.Create(() => new DeploymentManagerActor(
|
||||
_storage,
|
||||
_compilationService,
|
||||
_sharedScriptLibrary,
|
||||
null,
|
||||
new SiteRuntimeOptions { InstanceSweepIntervalSeconds = 0 },
|
||||
NullLogger<DeploymentManagerActor>.Instance,
|
||||
dclManager: null,
|
||||
healthCollector: health,
|
||||
serviceProvider: null,
|
||||
loggerFactory: null,
|
||||
configFetcher: null)));
|
||||
AwaitCondition(() => health.ReportCount > 0, TimeSpan.FromSeconds(20), TimeSpan.FromMilliseconds(25));
|
||||
return dm;
|
||||
}
|
||||
|
||||
private static DeployArtifactsCommand SharedScripts(string deploymentId, IEnumerable<string> names) =>
|
||||
new(deploymentId,
|
||||
names.Select(n => new SharedScriptArtifact(n, $"return \"{n}\".Length + {n.GetHashCode()};", null, null)).ToList(),
|
||||
null, null, null, null, null, DateTimeOffset.UtcNow);
|
||||
|
||||
/// <summary>
|
||||
/// A slow apply (many distinct shared scripts to compile) followed immediately by a fast one
|
||||
/// carrying a new script Z. Serialised in arrival order the final set is the second
|
||||
/// command's, so Z is present in storage AND in the compiled library. Run concurrently, the
|
||||
/// slow apply's delete-except ran after Z was stored and deleted it.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task OverlappingArtifactApplies_DoNotDeleteAnArtifactTheLaterApplyShipped()
|
||||
{
|
||||
var dm = CreateBootedDeploymentManager();
|
||||
|
||||
var slowSet = Enumerable.Range(0, 40).Select(i => $"Slow{i}").ToList();
|
||||
dm.Tell(SharedScripts("dep-A", slowSet));
|
||||
dm.Tell(SharedScripts("dep-B", ["Z"]));
|
||||
|
||||
var first = ExpectMsg<ArtifactDeploymentResponse>(TimeSpan.FromSeconds(60));
|
||||
var second = ExpectMsg<ArtifactDeploymentResponse>(TimeSpan.FromSeconds(60));
|
||||
Assert.True(first.Success, first.ErrorMessage);
|
||||
Assert.True(second.Success, second.ErrorMessage);
|
||||
|
||||
// Arrival order is apply order, and replies follow it.
|
||||
Assert.Equal("dep-A", first.DeploymentId);
|
||||
Assert.Equal("dep-B", second.DeploymentId);
|
||||
|
||||
var stored = (await _storage.GetAllSharedScriptsAsync()).Select(s => s.Name).ToList();
|
||||
Assert.Equal(["Z"], stored);
|
||||
Assert.True(_sharedScriptLibrary.Contains("Z"));
|
||||
Assert.DoesNotContain(_sharedScriptLibrary.GetRegisteredScriptNames(), n => n.StartsWith("Slow"));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user