dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

perf(deploy): flatten-session caching, bulk DeploySiteAsync, paged management queries

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-14 21:14:22 -04:00
parent ee193cd2bb
commit 48b3c40a7f
59 changed files with 3051 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentManagerOptions.cs
+39
View File
@@ -17,4 +17,43 @@ public class DeploymentManagerOptions
/// <summary>Timeout for acquiring an operation lock on an instance.</summary>
public TimeSpan OperationLockTimeout { get; set; } = TimeSpan.FromSeconds(5);
/// <summary>
/// Maximum number of instance deployments whose SITE ROUND-TRIP runs
/// concurrently during a bulk <c>DeploySiteAsync</c>. The flatten/validate and
/// persistence phases stay serial regardless (they share one non-thread-safe
/// <c>DbContext</c>); only the network wait is fanned out.
///
/// <para>
/// Bounded rather than unbounded so a large site cannot open one in-flight
/// gRPC command per instance against a two-node site pair. Four mirrors the
/// conservative end of the site's own apply concurrency; raise it only with
/// evidence from the target-scale load test.
/// </para>
/// </summary>
public int SiteDeploymentMaxParallelism { get; set; } = 4;
/// <summary>
/// Per-instance deadline applied to the site round-trip inside a bulk
/// <c>DeploySiteAsync</c>. One unreachable or wedged instance is recorded as a
/// failed row and the rest of the batch continues — the bulk operation never
/// inherits a single instance's hang.
/// </summary>
public TimeSpan SiteDeploymentTimeoutPerInstance { get; set; } = TimeSpan.FromSeconds(120);
/// <summary>
/// Retention window for TERMINAL deployment records (<c>Success</c>,
/// <c>Failed</c>). Deployments are insert-only — one row per attempt, forever —
/// so without a window the table grows without bound and every unfiltered
/// deployment query gets slower with the age of the system.
///
/// <para>
/// Deliberately generous (one year) because deployment history is
/// operator-facing forensics, and deliberately terminal-only: an
/// <c>InProgress</c> or <c>Pending</c> row is never purged regardless of age,
/// because it is exactly the row the query-before-redeploy reconciliation
/// path needs to find. Set to <see cref="TimeSpan.Zero"/> to disable purging.
/// </para>
/// </summary>
public TimeSpan TerminalDeploymentRecordRetention { get; set; } = TimeSpan.FromDays(365);
}
src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentManagerOptionsValidator.cs
+15
View File
@@ -27,5 +27,20 @@ public sealed class DeploymentManagerOptionsValidator : OptionsValidatorBase<Dep
builder.RequireThat(options.OperationLockTimeout > TimeSpan.Zero,
$"ScadaBridge:DeploymentManager:OperationLockTimeout must be a positive duration " +
$"(was {options.OperationLockTimeout}); it bounds acquiring an instance operation lock.");
builder.RequireThat(options.SiteDeploymentMaxParallelism > 0,
$"ScadaBridge:DeploymentManager:SiteDeploymentMaxParallelism must be at least 1 " +
$"(was {options.SiteDeploymentMaxParallelism}); it is the SemaphoreSlim bound on a bulk " +
$"site deployment's site round-trips, and a non-positive value would deadlock the fan-out.");
builder.RequireThat(options.SiteDeploymentTimeoutPerInstance > TimeSpan.Zero,
$"ScadaBridge:DeploymentManager:SiteDeploymentTimeoutPerInstance must be a positive duration " +
$"(was {options.SiteDeploymentTimeoutPerInstance}); it bounds each instance's site round-trip " +
$"inside a bulk site deployment.");
builder.RequireThat(options.TerminalDeploymentRecordRetention >= TimeSpan.Zero,
$"ScadaBridge:DeploymentManager:TerminalDeploymentRecordRetention must not be negative " +
$"(was {options.TerminalDeploymentRecordRetention}); use TimeSpan.Zero to disable purging of " +
$"terminal deployment records.");
}
}
src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentService.cs
+590 -64
View File
@@ -57,6 +57,24 @@ public class DeploymentService
/// </summary>
private const string TimeoutFailurePrefix = "Communication failure:";
/// <summary>Rows deleted per terminal-record purge batch.</summary>
private const int TerminalPurgeBatchSize = 500;
/// <summary>Batches deleted per opportunistic sweep, bounding one caller's latency cost.</summary>
private const int TerminalPurgeBatchesPerSweep = 4;
/// <summary>Minimum interval between opportunistic terminal-record sweeps.</summary>
private static readonly TimeSpan TerminalPurgeMinInterval = TimeSpan.FromHours(6);
/// <summary>
/// Ticks of the last opportunistic terminal-record sweep, process-wide.
/// <see cref="DeploymentService"/> is scoped, so this must be static to
/// rate-limit across requests; <see cref="Interlocked"/> makes the
/// claim-the-sweep step a compare-and-swap so concurrent readers do not all
/// sweep at once.
/// </summary>
private static long _lastTerminalPurgeTicks;
/// <summary>
/// Initializes a new instance of <see cref="DeploymentService"/> with all required dependencies.
/// </summary>
@@ -161,28 +179,113 @@ public class DeploymentService
int instanceId,
string user,
CancellationToken cancellationToken = default)
{
// Single-instance deploy = the same three phases the bulk path runs, with a
// batch of one and a private flatten session. Composing both entry points
// out of the same phase helpers is what keeps deployment identity,
// idempotency, lock coverage and optimistic concurrency from drifting apart
// between them.
var prepared = await PrepareDeploymentAsync(
instanceId, user, session: null, cancellationToken);
if (prepared.EarlyResult is { } early)
return early;
using (prepared.LockHandle)
{
var outcome = await SendDeploymentAsync(prepared, cancellationToken);
return await FinalizeDeploymentAsync(prepared, outcome, user, cancellationToken);
}
}
/// <summary>
/// Phase 1 of a deployment (serial, database-bound): validate the state
/// transition, take the per-instance operation lock, mint the deployment id,
/// flatten + validate, run query-before-redeploy reconciliation, stage the
/// <c>PendingDeployment</c> row and insert the <c>InProgress</c>
/// <see cref="DeploymentRecord"/>.
///
/// <para>
/// Everything here touches the scoped, non-thread-safe <c>DbContext</c>, so it
/// must run serially — a bulk site deploy loops this phase before fanning out
/// the phase-2 network waits. On success the returned
/// <see cref="PreparedDeployment"/> owns the operation lock; the CALLER is
/// responsible for disposing it once phase 3 has completed.
/// </para>
/// </summary>
/// <param name="instanceId">Instance being deployed.</param>
/// <param name="user">User attributed on the deployment record and audit rows.</param>
/// <param name="session">
/// Shared flatten session for a batch, or <see langword="null"/> for a private
/// one. Sharing it across a batch is what collapses N identical template-chain
/// walks into one.
/// </param>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>
/// A prepared deployment ready to send, or one carrying
/// <see cref="PreparedDeployment.EarlyResult"/> when the deploy resolved before
/// any site round-trip (validation failure, or a reconciled prior deployment).
/// </returns>
private async Task<PreparedDeployment> PrepareDeploymentAsync(
int instanceId,
string user,
FlattenSession? session,
CancellationToken cancellationToken)
{
// Load instance
var instance = await _repository.GetInstanceByIdAsync(instanceId, cancellationToken);
if (instance == null)
return Result<DeploymentRecord>.Failure($"Instance with ID {instanceId} not found.");
return PreparedDeployment.Resolved(Result<DeploymentRecord>.Failure($"Instance with ID {instanceId} not found."));
// Validate state transition
var transitionError = StateTransitionValidator.ValidateTransition(instance.State, "deploy");
if (transitionError != null)
return Result<DeploymentRecord>.Failure(transitionError);
return PreparedDeployment.Resolved(Result<DeploymentRecord>.Failure(transitionError));
// Acquire per-instance operation lock
using var lockHandle = await _lockManager.AcquireAsync(
var lockHandle = await _lockManager.AcquireAsync(
instance.UniqueName, _options.OperationLockTimeout, cancellationToken);
try
{
return await PrepareUnderLockAsync(instance, user, session, lockHandle, cancellationToken);
}
catch
{
// The lock is handed to the caller only on the success path; any fault
// during preparation must release it here or a failed prepare would
// wedge the instance until the process restarts.
lockHandle.Dispose();
throw;
}
}
/// <summary>
/// The body of <see cref="PrepareDeploymentAsync"/> that runs while the
/// per-instance operation lock is held. Split out so the lock can be released
/// on every fault path with one try/catch rather than a nested one per step.
/// </summary>
private async Task<PreparedDeployment> PrepareUnderLockAsync(
Instance instance,
string user,
FlattenSession? session,
IDisposable lockHandle,
CancellationToken cancellationToken)
{
var instanceId = instance.Id;
// Generate unique deployment ID
var deploymentId = Guid.NewGuid().ToString("N");
// Flatten configuration (captures template state at this point in time)
var flattenResult = await _flatteningPipeline.FlattenAndValidateAsync(instanceId, cancellationToken);
var flattenResult = await _flatteningPipeline.FlattenAndValidateAsync(
instanceId, cancellationToken, validateScripts: true, session);
if (flattenResult.IsFailure)
return Result<DeploymentRecord>.Failure($"Validation failed: {flattenResult.Error}");
{
lockHandle.Dispose();
return PreparedDeployment.Resolved(
Result<DeploymentRecord>.Failure($"Validation failed: {flattenResult.Error}"));
}
var flattenedConfig = flattenResult.Value.Configuration;
var revisionHash = flattenResult.Value.RevisionHash;
@@ -200,8 +303,9 @@ public class DeploymentService
validationResult.Errors.Count,
string.Join("; ", validationResult.Errors.Select(e => e.Message)));
return Result<DeploymentRecord>.Failure(
$"Pre-deployment validation failed: {validationResult.SummarizeErrors()}");
lockHandle.Dispose();
return PreparedDeployment.Resolved(Result<DeploymentRecord>.Failure(
$"Pre-deployment validation failed: {validationResult.SummarizeErrors()}"));
}
// Serialize for transmission (also the payload stored in the deployed
@@ -218,7 +322,10 @@ public class DeploymentService
var reconciled = await TryReconcileWithSiteAsync(
instance, revisionHash, configJson, user, cancellationToken);
if (reconciled != null)
return Result<DeploymentRecord>.Success(reconciled);
{
lockHandle.Dispose();
return PreparedDeployment.Resolved(Result<DeploymentRecord>.Success(reconciled));
}
// Notify-and-fetch: the site fetches the staged config from
// CentralFetchBaseUrl, so a deploy is impossible without it. Fail fast
@@ -227,8 +334,11 @@ public class DeploymentService
// confusing downstream site-fetch failure (and no InProgress record is
// stranded).
if (string.IsNullOrEmpty(_commOptions.CentralFetchBaseUrl))
return Result<DeploymentRecord>.Failure(
"CentralFetchBaseUrl is not configured — required for deployment (notify-and-fetch).");
{
lockHandle.Dispose();
return PreparedDeployment.Resolved(Result<DeploymentRecord>.Failure(
"CentralFetchBaseUrl is not configured — required for deployment (notify-and-fetch)."));
}
// Create the deployment record directly in InProgress.
//
@@ -275,12 +385,8 @@ public class DeploymentService
deploymentId, instance.UniqueName, revisionHash, user, stagedAt,
_commOptions.CentralFetchBaseUrl, token);
_logger.LogInformation(
"Sending deployment {DeploymentId} for instance {Instance} to site {SiteId} (notify-and-fetch)",
deploymentId, instance.UniqueName, siteId);
// Cleanup of the staged PendingDeployment is TTL-based ONLY — the row
// is deliberately NOT deleted on success or in the catch. On a
// is deliberately NOT deleted on success or on failure. On a
// central-side Ask timeout the site may have applied AND told the
// standby node to fetch; deleting now would 404 that in-flight
// standby fetch and break failover. Supersession bounds pending rows
@@ -288,8 +394,89 @@ public class DeploymentService
// leaving rows for TTL purge is safe. Expired rows are swept by the
// central PendingDeploymentPurgeActor singleton on its
// CommunicationOptions.PendingDeploymentPurgeInterval cadence.
var response = await _communicationService.RefreshDeploymentAsync(siteId, command, cancellationToken);
return new PreparedDeployment(
instance, deploymentId, record, revisionHash, configJson,
siteId, command, lockHandle, EarlyResult: null);
}
catch (Exception ex)
{
// Staging failed before anything was sent. Record the failure exactly
// as the post-send path does (never leave the record InProgress) and
// release the lock — there is no phase 2 or 3 to run.
await MarkDeploymentFailedAsync(record, instance, deploymentId, user, ex);
lockHandle.Dispose();
return PreparedDeployment.Resolved(FailureResultFor(ex));
}
}
/// <summary>
/// Phase 2 of a deployment (parallelisable, network-bound): the
/// <c>RefreshDeploymentCommand</c> round-trip to the site. Touches no
/// repository and mutates no shared state, which is what makes it safe to fan
/// out across a batch while phases 1 and 3 stay serial.
/// </summary>
/// <param name="prepared">The prepared deployment to send.</param>
/// <param name="cancellationToken">Cancellation token, already carrying any per-instance deadline.</param>
/// <returns>The site's response, or the exception that prevented one.</returns>
private async Task<SendOutcome> SendDeploymentAsync(
PreparedDeployment prepared,
CancellationToken cancellationToken)
{
try
{
_logger.LogInformation(
"Sending deployment {DeploymentId} for instance {Instance} to site {SiteId} (notify-and-fetch)",
prepared.DeploymentId, prepared.Instance.UniqueName, prepared.SiteIdentifier);
var response = await _communicationService.RefreshDeploymentAsync(
prepared.SiteIdentifier, prepared.Command, cancellationToken);
return new SendOutcome(response, null);
}
catch (Exception ex)
{
return new SendOutcome(null, ex);
}
}
/// <summary>
/// Phase 3 of a deployment (serial, database-bound): commit the terminal
/// status, apply post-success side effects, and write the audit row.
/// </summary>
/// <param name="prepared">The prepared deployment whose send just completed.</param>
/// <param name="outcome">The site response or the fault from phase 2.</param>
/// <param name="user">User attributed on the audit rows.</param>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>The deployment result for the caller.</returns>
private async Task<Result<DeploymentRecord>> FinalizeDeploymentAsync(
PreparedDeployment prepared,
SendOutcome outcome,
string user,
CancellationToken cancellationToken)
{
var instance = prepared.Instance;
var record = prepared.Record;
var deploymentId = prepared.DeploymentId;
var instanceId = instance.Id;
if (outcome.Error is { } sendError)
{
// Any fault out of the site round-trip (timeout, cancellation,
// transport, serialization) must leave the deployment record as
// Failed -- the design requires an interrupted deployment to be
// treated as failed, never stuck in InProgress.
await MarkDeploymentFailedAsync(record, instance, deploymentId, user, sendError);
_logger.LogError(sendError,
"Deployment {DeploymentId} for instance {Instance} failed",
deploymentId, instance.UniqueName);
return FailureResultFor(sendError);
}
var response = outcome.Response!;
try
{
// Update status based on site response.
record.Status = response.Status;
record.ErrorMessage = response.ErrorMessage;
@@ -322,7 +509,7 @@ public class DeploymentService
// logged loudly for operator reconciliation but must not flip
// the already-committed Success record back to Failed.
await ApplyPostSuccessSideEffectsAsync(
instance, deploymentId, revisionHash, configJson,
instance, deploymentId, prepared.RevisionHash, prepared.ConfigJson,
forceEnabledState: true, cancellationToken);
}
@@ -345,61 +532,366 @@ public class DeploymentService
}
catch (Exception ex)
{
// Any exception out of the try (timeout,
// cancellation, transport, serialization, DB) must leave the
// deployment record as Failed -- the design requires an interrupted
// deployment to be treated as failed, never stuck in InProgress.
//
// The failure-status write must NOT use the
// operation's cancellation token. If the operation was cancelled or
// timed out, that token is already cancelled and the cleanup writes
// would themselves throw before the Failed status is persisted.
// Use CancellationToken.None so the failure is durably recorded.
var isTimeout = ex is TimeoutException or OperationCanceledException or Akka.Actor.AskTimeoutException;
record.Status = DeploymentStatus.Failed;
record.ErrorMessage = isTimeout
? $"{TimeoutFailurePrefix} {ex.Message}"
: $"Deployment error: {ex.Message}";
record.CompletedAt = DateTimeOffset.UtcNow;
try
{
await _repository.UpdateDeploymentRecordAsync(record, CancellationToken.None);
// Note: if the staging SaveChangesAsync above was interrupted, an
// Added PendingDeployment may still be tracked and will be
// committed by this cleanup save. That row is orphaned (no
// RefreshDeploymentCommand was sent, so no site holds its token)
// and is removed by TTL purge / superseded by the next deploy --
// harmless.
await _repository.SaveChangesAsync(CancellationToken.None);
NotifyStatusChange(record);
await _auditService.LogAsync(user, "DeployFailed", "Instance", instanceId.ToString(),
instance.UniqueName, new { DeploymentId = deploymentId, Error = ex.Message },
CancellationToken.None);
}
catch (Exception cleanupEx)
{
// The deployment already failed; a failed cleanup write must not
// mask the original error. Log loudly so an operator can reconcile.
_logger.LogError(cleanupEx,
"Failed to persist Failed status for deployment {DeploymentId} of instance {Instance} " +
"after deployment error: {Error}",
deploymentId, instance.UniqueName, ex.Message);
}
// A fault in the post-send persistence path is handled identically to
// a send fault: the record must never be left InProgress.
await MarkDeploymentFailedAsync(record, instance, deploymentId, user, ex);
_logger.LogError(ex,
"Deployment {DeploymentId} for instance {Instance} failed",
deploymentId, instance.UniqueName);
return Result<DeploymentRecord>.Failure(
isTimeout
? $"Deployment timed out: {ex.Message}"
: $"Deployment failed: {ex.Message}");
return FailureResultFor(ex);
}
}
/// <summary>
/// Writes the terminal <see cref="DeploymentStatus.Failed"/> status (plus the
/// failure audit row) for a deployment that faulted, on any phase.
///
/// <para>
/// The failure-status write deliberately uses <see cref="CancellationToken.None"/>:
/// if the operation was cancelled or timed out, the operation's own token is
/// already cancelled and these cleanup writes would themselves throw before
/// the Failed status was persisted, leaving the record stuck InProgress.
/// </para>
///
/// <para>
/// A fault DURING cleanup is logged loudly and swallowed — it must not mask
/// the original error the caller is about to report.
/// </para>
/// </summary>
private async Task MarkDeploymentFailedAsync(
DeploymentRecord record,
Instance instance,
string deploymentId,
string user,
Exception cause)
{
record.Status = DeploymentStatus.Failed;
record.ErrorMessage = IsTimeoutFault(cause)
? $"{TimeoutFailurePrefix} {cause.Message}"
: $"Deployment error: {cause.Message}";
record.CompletedAt = DateTimeOffset.UtcNow;
try
{
await _repository.UpdateDeploymentRecordAsync(record, CancellationToken.None);
// Note: if the staging SaveChangesAsync was interrupted, an Added
// PendingDeployment may still be tracked and will be committed by this
// cleanup save. That row is orphaned (no RefreshDeploymentCommand was
// sent, so no site holds its token) and is removed by TTL purge /
// superseded by the next deploy -- harmless.
await _repository.SaveChangesAsync(CancellationToken.None);
NotifyStatusChange(record);
await _auditService.LogAsync(user, "DeployFailed", "Instance", instance.Id.ToString(),
instance.UniqueName, new { DeploymentId = deploymentId, Error = cause.Message },
CancellationToken.None);
}
catch (Exception cleanupEx)
{
_logger.LogError(cleanupEx,
"Failed to persist Failed status for deployment {DeploymentId} of instance {Instance} " +
"after deployment error: {Error}",
deploymentId, instance.UniqueName, cause.Message);
}
}
/// <summary>
/// True when a deployment fault is a timeout/cancellation rather than a hard
/// error. Drives both the <see cref="TimeoutFailurePrefix"/> marker (which the
/// query-before-redeploy trigger reads) and the caller-facing wording.
/// </summary>
private static bool IsTimeoutFault(Exception ex) =>
ex is TimeoutException or OperationCanceledException or Akka.Actor.AskTimeoutException;
/// <summary>Maps a deployment fault to the caller-facing failure result.</summary>
private static Result<DeploymentRecord> FailureResultFor(Exception ex) =>
Result<DeploymentRecord>.Failure(
IsTimeoutFault(ex)
? $"Deployment timed out: {ex.Message}"
: $"Deployment failed: {ex.Message}");
/// <summary>
/// A deployment that has cleared phase 1 and is ready for its site round-trip,
/// OR one that resolved during phase 1 (carrying <see cref="EarlyResult"/>).
///
/// <para>
/// When <see cref="EarlyResult"/> is <see langword="null"/> this record OWNS the
/// per-instance operation lock in <see cref="LockHandle"/>, and the caller must
/// dispose it after phase 3. When <see cref="EarlyResult"/> is set the lock has
/// already been released and every other member is a placeholder.
/// </para>
/// </summary>
private sealed record PreparedDeployment(
Instance Instance,
string DeploymentId,
DeploymentRecord Record,
string RevisionHash,
string ConfigJson,
string SiteIdentifier,
RefreshDeploymentCommand Command,
IDisposable? LockHandle,
Result<DeploymentRecord>? EarlyResult)
{
/// <summary>Builds a phase-1-resolved deployment carrying the given result and holding no lock.</summary>
/// <param name="result">The result to hand back to the caller.</param>
/// <returns>A prepared deployment whose <see cref="EarlyResult"/> is set.</returns>
public static PreparedDeployment Resolved(Result<DeploymentRecord> result) =>
new(null!, string.Empty, null!, string.Empty, string.Empty, string.Empty, null!, null, result);
}
/// <summary>
/// Result of phase 2: either the site's response or the exception that
/// prevented one. Exactly one is non-null.
/// </summary>
private readonly record struct SendOutcome(DeploymentStatusResponse? Response, Exception? Error);
/// <summary>
/// Deploy every deployable instance at a site in one operation.
///
/// <para>
/// Structured as the three deployment phases with the middle one fanned out:
/// </para>
/// <list type="number">
/// <item>
/// <b>Prepare (serial).</b> Every instance is flattened, validated, staged
/// and given an <c>InProgress</c> record on the caller's single scoped
/// <c>DbContext</c>. All instances share ONE <see cref="FlattenSession"/>,
/// so a template chain common to N instances is walked once, and the
/// session-global queries (shared scripts, schema library, the site's data
/// connections) run once for the whole batch instead of once per instance.
/// </item>
/// <item>
/// <b>Send (bounded parallel).</b> Site round-trips run concurrently up to
/// <see cref="DeploymentManagerOptions.SiteDeploymentMaxParallelism"/>,
/// each under its own
/// <see cref="DeploymentManagerOptions.SiteDeploymentTimeoutPerInstance"/>
/// deadline, so one wedged instance cannot stall the batch. This phase
/// touches no repository — that is precisely why it is the only phase that
/// may run in parallel against a non-thread-safe <c>DbContext</c>.
/// </item>
/// <item>
/// <b>Finalize (serial).</b> Terminal statuses, post-success side effects
/// and audit rows are committed one instance at a time, then each
/// instance's operation lock is released.
/// </item>
/// </list>
///
/// <para>
/// Semantics per instance are byte-for-byte those of
/// <see cref="DeployInstanceAsync"/> — its own deployment id and revision hash,
/// its own operation lock held across the whole operation, its own
/// optimistically-concurrent status record, and the same query-before-redeploy
/// idempotency check. Instances that cannot be deployed (wrong state, failed
/// validation, lock already held) are reported as failed rows; the batch is
/// NOT all-or-nothing, mirroring the artifact deployment path.
/// </para>
/// </summary>
/// <param name="siteId">The database ID of the site whose instances are deployed.</param>
/// <param name="user">The username initiating the deployment, recorded on every record and audit row.</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>A per-instance result matrix, or a failure result when the site itself cannot be resolved.</returns>
public async Task<Result<SiteDeploymentSummary>> DeploySiteAsync(
int siteId,
string user,
CancellationToken cancellationToken = default)
{
var site = await _siteRepository.GetSiteByIdAsync(siteId, cancellationToken);
if (site == null)
return Result<SiteDeploymentSummary>.Failure($"Site with ID {siteId} not found.");
var instances = await _siteRepository.GetInstancesBySiteIdAsync(siteId, cancellationToken);
if (instances.Count == 0)
return Result<SiteDeploymentSummary>.Success(new SiteDeploymentSummary(site.SiteIdentifier, [], 0, 0));
// ---- Phase 1: prepare, serially, on ONE shared flatten session. ----
var session = _flatteningPipeline.CreateSession();
var prepared = new List<PreparedDeployment>(instances.Count);
var results = new List<InstanceDeploymentResult>(instances.Count);
foreach (var instance in instances)
{
cancellationToken.ThrowIfCancellationRequested();
PreparedDeployment step;
try
{
step = await PrepareDeploymentAsync(instance.Id, user, session, cancellationToken);
}
catch (Exception ex)
{
// A prepare fault (most commonly a TimeoutException from the
// operation lock because another operation holds this instance)
// fails only this instance. Recording it and moving on is what
// makes a bulk deploy usable while individual instances are busy.
_logger.LogWarning(ex,
"Preparing instance {Instance} for bulk deployment of site {SiteId} failed",
instance.UniqueName, site.SiteIdentifier);
results.Add(new InstanceDeploymentResult(
instance.Id, instance.UniqueName, null, false, ex.Message));
continue;
}
if (step.EarlyResult is { } early)
{
results.Add(ToInstanceResult(instance.Id, instance.UniqueName, early));
continue;
}
prepared.Add(step);
}
// ---- Phase 2: bounded-parallel site round-trips. ----
var outcomes = await SendPreparedAsync(prepared, cancellationToken);
// ---- Phase 3: finalize, serially, releasing each lock as we go. ----
for (var i = 0; i < prepared.Count; i++)
{
var step = prepared[i];
try
{
var result = await FinalizeDeploymentAsync(step, outcomes[i], user, cancellationToken);
results.Add(ToInstanceResult(step.Instance.Id, step.Instance.UniqueName, result, step.DeploymentId));
}
finally
{
step.LockHandle?.Dispose();
}
}
var summary = new SiteDeploymentSummary(
site.SiteIdentifier,
results,
results.Count(r => r.Success),
results.Count(r => !r.Success));
_logger.LogInformation(
"Bulk deployment of site {SiteId} complete: {SuccessCount} succeeded, {FailureCount} failed " +
"({ChainLoads} template-chain load(s) across {InstanceCount} instance(s))",
site.SiteIdentifier, summary.SuccessCount, summary.FailureCount,
session.ChainLoads, instances.Count);
return Result<SiteDeploymentSummary>.Success(summary);
}
/// <summary>
/// Runs phase 2 for a whole batch: every prepared deployment's site round-trip,
/// concurrent up to <see cref="DeploymentManagerOptions.SiteDeploymentMaxParallelism"/>
/// and each bounded by
/// <see cref="DeploymentManagerOptions.SiteDeploymentTimeoutPerInstance"/>.
/// Outcomes are returned positionally so phase 3 can pair them back with their
/// prepared deployment.
/// </summary>
private async Task<SendOutcome[]> SendPreparedAsync(
IReadOnlyList<PreparedDeployment> prepared,
CancellationToken cancellationToken)
{
var outcomes = new SendOutcome[prepared.Count];
if (prepared.Count == 0)
return outcomes;
using var gate = new SemaphoreSlim(_options.SiteDeploymentMaxParallelism);
var sends = prepared.Select(async (step, index) =>
{
await gate.WaitAsync(cancellationToken);
try
{
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cts.CancelAfter(_options.SiteDeploymentTimeoutPerInstance);
outcomes[index] = await SendDeploymentAsync(step, cts.Token);
}
finally
{
gate.Release();
}
});
await Task.WhenAll(sends);
return outcomes;
}
/// <summary>Maps a per-instance deployment result into the bulk summary row shape.</summary>
private static InstanceDeploymentResult ToInstanceResult(
int instanceId,
string uniqueName,
Result<DeploymentRecord> result,
string? deploymentId = null) =>
result.IsSuccess
? new InstanceDeploymentResult(
instanceId, uniqueName, result.Value.DeploymentId, true, null)
: new InstanceDeploymentResult(
instanceId, uniqueName, deploymentId, false, result.Error);
/// <summary>
/// Opportunistic retention sweep for TERMINAL deployment records, rate-limited
/// to once every <see cref="TerminalPurgeMinInterval"/> per process and bounded
/// to <see cref="TerminalPurgeBatchesPerSweep"/> batches per call.
///
/// <para>
/// Swept on read (the deployment-list path) rather than from a timer for the
/// same reason expired secured writes are: the component owns no scheduling
/// infrastructure of its own, and the read path is exactly the operation that
/// suffers when the table is allowed to grow without bound. Both central nodes
/// may sweep; the delete is idempotent, so a duplicate pass is a no-op.
/// </para>
///
/// <para>
/// Best-effort: a failed sweep is logged and swallowed. Retention maintenance
/// must never fail the operator's actual query.
/// </para>
/// </summary>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>The number of records purged by this call (0 when the sweep was skipped).</returns>
public async Task<int> TryPurgeTerminalDeploymentRecordsAsync(CancellationToken cancellationToken = default)
{
var retention = _options.TerminalDeploymentRecordRetention;
if (retention <= TimeSpan.Zero)
return 0;
var now = DateTimeOffset.UtcNow;
var last = Interlocked.Read(ref _lastTerminalPurgeTicks);
if (last != 0 && now.UtcTicks - last < TerminalPurgeMinInterval.Ticks)
return 0;
// Claim the sweep. A loser of this CAS returns immediately rather than
// running a redundant second pass.
if (Interlocked.CompareExchange(ref _lastTerminalPurgeTicks, now.UtcTicks, last) != last)
return 0;
var cutoff = now - retention;
var purged = 0;
try
{
for (var batch = 0; batch < TerminalPurgeBatchesPerSweep; batch++)
{
var deleted = await _repository.PurgeTerminalDeploymentRecordsAsync(
cutoff, TerminalPurgeBatchSize, cancellationToken);
purged += deleted;
if (deleted < TerminalPurgeBatchSize)
break;
}
if (purged > 0)
{
_logger.LogInformation(
"Purged {Count} terminal deployment record(s) completed before {Cutoff} " +
"(retention {Retention}).",
purged, cutoff, retention);
}
}
catch (Exception ex)
{
_logger.LogWarning(ex,
"Terminal deployment-record retention sweep failed; retrying after {Interval}.",
TerminalPurgeMinInterval);
}
return purged;
}
/// <summary>
/// Disable an instance. Stops Instance Actor, retains config, S&amp;F drains.
/// </summary>
@@ -1178,3 +1670,37 @@ public record DeploymentComparisonResult(
bool IsStale,
DateTimeOffset DeployedAt,
ConfigurationDiff? Diff = null);
/// <summary>
/// One instance's outcome inside a bulk site deployment.
/// </summary>
/// <param name="InstanceId">Database id of the instance.</param>
/// <param name="UniqueName">The instance's unique name (also its operation-lock key).</param>
/// <param name="DeploymentId">
/// The deployment id minted for this instance, or <see langword="null"/> when the
/// instance never got that far (state-transition rejection, lock contention).
/// </param>
/// <param name="Success">Whether the site confirmed the apply.</param>
/// <param name="ErrorMessage">Failure detail; <see langword="null"/> on success.</param>
public record InstanceDeploymentResult(
int InstanceId,
string UniqueName,
string? DeploymentId,
bool Success,
string? ErrorMessage);
/// <summary>
/// Result matrix of a bulk site deployment. Mirrors the shape of
/// <c>ArtifactDeploymentSummary</c>: successes are not rolled back when other
/// instances fail, and each failed instance is individually retryable through the
/// ordinary single-instance deploy.
/// </summary>
/// <param name="SiteIdentifier">The site's string identifier (e.g. <c>site-a</c>).</param>
/// <param name="InstanceResults">Per-instance outcomes.</param>
/// <param name="SuccessCount">Number of instances the site confirmed.</param>
/// <param name="FailureCount">Number of instances that failed for any reason.</param>
public record SiteDeploymentSummary(
string SiteIdentifier,
IReadOnlyList<InstanceDeploymentResult> InstanceResults,
int SuccessCount,
int FailureCount);
src/ZB.MOM.WW.ScadaBridge.DeploymentManager/FlattenSession.cs
+257
View File
@@ -0,0 +1,257 @@
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Templates;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
namespace ZB.MOM.WW.ScadaBridge.DeploymentManager;
/// <summary>
/// Memoises the repository reads that <see cref="FlatteningPipeline"/> performs,
/// for the lifetime of ONE flatten/validate session.
///
/// <para>
/// Without a session every instance flatten re-walks its whole template
/// inheritance chain (one query per link), re-loads the compositions of every
/// template in that chain and of every composed chain it reaches, and re-issues
/// the three session-global queries — shared scripts, the shared-schema library,
/// and the target site's data connections. For a bulk deploy of N instances off
/// the same template that is N× the same work; even a single instance re-loads a
/// composed chain once per composing template.
/// </para>
///
/// <para>
/// <b>Cache key.</b> Templates are keyed on <c>(id, ITemplateGraphWatermark
/// version)</c> and chain MEMBERSHIP additionally on the watermark's
/// <see cref="ITemplateGraphWatermark.StructureVersion"/>, so a template edited
/// mid-session is re-read rather than served from the memo. The session is also
/// short-lived by construction — it is created per deploy/validate operation and
/// discarded with it — so a cached graph can never outlive the operation that
/// captured it. That is the design's "template state is captured at the time of
/// flatten" guarantee, unchanged: the session narrows the capture window, it does
/// not widen it.
/// </para>
///
/// <para>
/// <b>Threading.</b> A session is NOT thread-safe and must not be shared across
/// threads. It caches entities materialised by one scoped <c>DbContext</c>, which
/// is itself single-threaded, and it hands out tracked entities that the
/// in-flight bundle importer relies on observing. <c>DeploySiteAsync</c> honours
/// this by running the whole prepare (flatten) phase serially on one scope and
/// parallelising only the site round-trips, which touch no repository at all.
/// </para>
/// </summary>
public sealed class FlattenSession
{
private readonly ITemplateGraphWatermark _watermark;
private readonly Dictionary<int, (long Version, Template Template)> _templates = new();
private readonly Dictionary<int, CachedChain> _chains = new();
private readonly Dictionary<int, (long Version, IReadOnlyList<TemplateComposition> Compositions)> _compositions = new();
private readonly Dictionary<int, IReadOnlyDictionary<int, DataConnection>> _dataConnections = new();
private IReadOnlyList<ResolvedScript>? _sharedScripts;
private IReadOnlyDictionary<string, string>? _schemaLibrary;
/// <summary>Initializes a session bound to the process-wide graph watermark.</summary>
/// <param name="watermark">Watermark supplying the per-template versions used as cache keys.</param>
public FlattenSession(ITemplateGraphWatermark watermark)
{
_watermark = watermark ?? throw new ArgumentNullException(nameof(watermark));
}
/// <summary>
/// Number of template-chain walks that actually hit the repository. Reset
/// never; used by tests to assert that N instances sharing a template load
/// that chain exactly once.
/// </summary>
public int ChainLoads { get; private set; }
/// <summary>Number of single-template repository reads that missed the memo.</summary>
public int TemplateLoads { get; private set; }
/// <summary>Number of per-template composition queries that missed the memo.</summary>
public int CompositionLoads { get; private set; }
/// <summary>Number of times the session-global queries (shared scripts, schema library) ran.</summary>
public int GlobalLoads { get; private set; }
/// <summary>Number of per-site data-connection queries that missed the memo.</summary>
public int DataConnectionLoads { get; private set; }
/// <summary>
/// Every template id this session resolved — the union of all inheritance and
/// composed chains it walked. The staleness fast path records the watermark
/// version of exactly this set, so a change to ANY template that fed a
/// flatten invalidates the memoised revision hash.
/// </summary>
public IReadOnlyCollection<int> VisitedTemplateIds => _templates.Keys;
/// <summary>
/// Returns the inheritance chain for <paramref name="templateId"/> (the
/// template itself, then each ancestor), loading it through
/// <paramref name="loadTemplate"/> on a miss.
/// </summary>
/// <param name="templateId">Root of the chain to build.</param>
/// <param name="loadTemplate">Repository read for one template with its children.</param>
/// <returns>The chain, ordered leaf-first.</returns>
public async Task<IReadOnlyList<Template>> GetChainAsync(
int templateId,
Func<int, Task<Template?>> loadTemplate)
{
var structure = _watermark.StructureVersion;
if (_chains.TryGetValue(templateId, out var cached) && IsChainCurrent(cached, structure))
return cached.Chain;
var chain = new List<Template>();
var memberVersions = new Dictionary<int, long>();
var currentId = (int?)templateId;
// Guard against a cyclic parent edge. Acyclicity is enforced on save, but
// this walk must not hang if a bad row ever reaches the database.
var seen = new HashSet<int>();
while (currentId.HasValue && seen.Add(currentId.Value))
{
memberVersions[currentId.Value] = _watermark.GetTemplateVersion(currentId.Value);
var template = await GetTemplateAsync(currentId.Value, loadTemplate);
if (template == null) break;
chain.Add(template);
currentId = template.ParentTemplateId;
}
ChainLoads++;
_chains[templateId] = new CachedChain(structure, memberVersions, chain);
return chain;
}
/// <summary>
/// A memoised chain plus the exact watermark readings it was built under.
/// </summary>
/// <param name="Structure">Structure version, covering chain MEMBERSHIP.</param>
/// <param name="MemberVersions">Per-template version of every link, covering chain CONTENTS.</param>
/// <param name="Chain">The chain itself, ordered leaf-first.</param>
private sealed record CachedChain(
long Structure,
IReadOnlyDictionary<int, long> MemberVersions,
IReadOnlyList<Template> Chain);
/// <summary>
/// A cached chain is current only when BOTH the graph shape and every member's
/// own version are unchanged.
///
/// <para>
/// Both halves are load-bearing. Structure alone would miss an ordinary member
/// edit (an attribute added to a template already in the chain), which changes
/// what the flattener produces without changing which templates it walks.
/// Member versions alone would miss a re-parent, which changes membership
/// without touching any surviving member. Checking only one would let a session
/// serve a chain that no longer reflects the database.
/// </para>
/// </summary>
private bool IsChainCurrent(CachedChain cached, long structure)
{
if (cached.Structure != structure)
return false;
foreach (var (templateId, version) in cached.MemberVersions)
{
if (_watermark.GetTemplateVersion(templateId) != version)
return false;
}
return true;
}
/// <summary>
/// Returns one template with its children, loading it through
/// <paramref name="loadTemplate"/> on a miss or when its watermark version moved.
/// </summary>
/// <param name="templateId">Template to read.</param>
/// <param name="loadTemplate">Repository read for one template with its children.</param>
/// <returns>The template, or <see langword="null"/> when it does not exist.</returns>
public async Task<Template?> GetTemplateAsync(int templateId, Func<int, Task<Template?>> loadTemplate)
{
var version = _watermark.GetTemplateVersion(templateId);
if (_templates.TryGetValue(templateId, out var cached) && cached.Version == version)
return cached.Template;
var template = await loadTemplate(templateId);
TemplateLoads++;
if (template != null)
_templates[templateId] = (version, template);
return template;
}
/// <summary>
/// Returns the composition rows declared directly on <paramref name="templateId"/>.
/// </summary>
/// <param name="templateId">Template whose compositions are read.</param>
/// <param name="loadCompositions">Repository read for one template's compositions.</param>
/// <returns>The composition rows.</returns>
public async Task<IReadOnlyList<TemplateComposition>> GetCompositionsAsync(
int templateId,
Func<int, Task<IReadOnlyList<TemplateComposition>>> loadCompositions)
{
var version = _watermark.GetTemplateVersion(templateId);
if (_compositions.TryGetValue(templateId, out var cached) && cached.Version == version)
return cached.Compositions;
var compositions = await loadCompositions(templateId);
CompositionLoads++;
_compositions[templateId] = (version, compositions);
return compositions;
}
/// <summary>
/// Returns the site's data connections keyed by id. Cached per site for the
/// session — a bulk site deploy resolves them once for the whole batch.
/// </summary>
/// <param name="siteId">Site whose connections are read.</param>
/// <param name="loadConnections">Repository read for a site's data connections.</param>
/// <returns>The site's connections keyed by database id.</returns>
public async Task<IReadOnlyDictionary<int, DataConnection>> GetDataConnectionsAsync(
int siteId,
Func<int, Task<IReadOnlyDictionary<int, DataConnection>>> loadConnections)
{
if (_dataConnections.TryGetValue(siteId, out var cached))
return cached;
var connections = await loadConnections(siteId);
DataConnectionLoads++;
_dataConnections[siteId] = connections;
return connections;
}
/// <summary>
/// Returns the shared-script set used for semantic validation. Hoisted to the
/// session so a bulk deploy issues this query once instead of once per instance.
/// </summary>
/// <param name="loadSharedScripts">Repository read for all shared scripts.</param>
/// <returns>The resolved shared scripts.</returns>
public async Task<IReadOnlyList<ResolvedScript>> GetSharedScriptsAsync(
Func<Task<IReadOnlyList<ResolvedScript>>> loadSharedScripts)
{
if (_sharedScripts != null)
return _sharedScripts;
_sharedScripts = await loadSharedScripts();
GlobalLoads++;
return _sharedScripts;
}
/// <summary>
/// Returns the shared JSON-Schema library as a name → schema-JSON map, hoisted
/// to the session for the same reason as the shared scripts.
/// </summary>
/// <param name="loadSchemaLibrary">Repository read building the schema library map.</param>
/// <returns>The schema library.</returns>
public async Task<IReadOnlyDictionary<string, string>> GetSchemaLibraryAsync(
Func<Task<IReadOnlyDictionary<string, string>>> loadSchemaLibrary)
{
if (_schemaLibrary != null)
return _schemaLibrary;
_schemaLibrary = await loadSchemaLibrary();
GlobalLoads++;
return _schemaLibrary;
}
}
src/ZB.MOM.WW.ScadaBridge.DeploymentManager/FlatteningPipeline.cs
+52 -34
View File
@@ -1,6 +1,7 @@
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
using ZB.MOM.WW.ScadaBridge.TemplateEngine.Flattening;
@@ -23,6 +24,7 @@ public class FlatteningPipeline : IFlatteningPipeline
private readonly ValidationService _validationService;
private readonly RevisionHashService _revisionHashService;
private readonly ISharedSchemaRepository _sharedSchemaRepo;
private readonly ITemplateGraphWatermark _watermark;
/// <summary>Initializes a new <see cref="FlatteningPipeline"/> with the required template engine and site repositories and services.</summary>
/// <param name="templateRepo">Repository for loading templates and instance data.</param>
@@ -35,13 +37,19 @@ public class FlatteningPipeline : IFlatteningPipeline
/// look up <c>lib:Name</c> library references so a dangling reference in any validated
/// script schema becomes a deploy-blocking error.
/// </param>
/// <param name="watermark">
/// Process-wide template version watermark. Supplies the per-template version
/// used as the <see cref="FlattenSession"/> cache-key discriminator, so a
/// memoised chain is dropped the moment any of its templates is edited.
/// </param>
public FlatteningPipeline(
ITemplateEngineRepository templateRepo,
ISiteRepository siteRepo,
FlatteningService flatteningService,
ValidationService validationService,
RevisionHashService revisionHashService,
ISharedSchemaRepository sharedSchemaRepo)
ISharedSchemaRepository sharedSchemaRepo,
ITemplateGraphWatermark watermark)
{
_templateRepo = templateRepo;
_siteRepo = siteRepo;
@@ -49,21 +57,35 @@ public class FlatteningPipeline : IFlatteningPipeline
_validationService = validationService;
_revisionHashService = revisionHashService;
_sharedSchemaRepo = sharedSchemaRepo;
_watermark = watermark;
}
/// <inheritdoc />
public FlattenSession CreateSession() => new(_watermark);
/// <inheritdoc />
public async Task<Result<FlatteningPipelineResult>> FlattenAndValidateAsync(
int instanceId,
CancellationToken cancellationToken = default,
bool validateScripts = true)
bool validateScripts = true,
FlattenSession? session = null)
{
// A caller that supplies no session gets a private one. That still pays
// off inside a single flatten: the composed-chain walk below reaches the
// same template through several composing parents, and the memo collapses
// those to one load each.
session ??= CreateSession();
// Load instance with full graph
var instance = await _templateRepo.GetInstanceByIdAsync(instanceId, cancellationToken);
if (instance == null)
return Result<FlatteningPipelineResult>.Failure($"Instance with ID {instanceId} not found.");
Task<Commons.Entities.Templates.Template?> LoadTemplate(int id) =>
_templateRepo.GetTemplateWithChildrenAsync(id, cancellationToken);
// Build template chain
var templateChain = await BuildTemplateChainAsync(instance.TemplateId, cancellationToken);
var templateChain = await session.GetChainAsync(instance.TemplateId, LoadTemplate);
if (templateChain.Count == 0)
return Result<FlatteningPipelineResult>.Failure("Template chain is empty.");
@@ -84,7 +106,9 @@ public class FlatteningPipeline : IFlatteningPipeline
{
if (!processedTemplateIds.Add(template.Id)) continue;
var compositions = await _templateRepo.GetCompositionsByTemplateIdAsync(template.Id, cancellationToken);
var compositions = await session.GetCompositionsAsync(
template.Id,
id => _templateRepo.GetCompositionsByTemplateIdAsync(id, cancellationToken));
if (compositions.Count == 0) continue;
compositionMap[template.Id] = compositions;
@@ -92,7 +116,7 @@ public class FlatteningPipeline : IFlatteningPipeline
{
if (composedChains.ContainsKey(comp.ComposedTemplateId)) continue;
var composedChain = await BuildTemplateChainAsync(comp.ComposedTemplateId, cancellationToken);
var composedChain = await session.GetChainAsync(comp.ComposedTemplateId, LoadTemplate);
composedChains[comp.ComposedTemplateId] = composedChain;
pendingChains.Enqueue(composedChain);
}
@@ -100,7 +124,9 @@ public class FlatteningPipeline : IFlatteningPipeline
}
// Load data connections for the site
var dataConnections = await LoadDataConnectionsAsync(instance.SiteId, cancellationToken);
var dataConnections = await session.GetDataConnectionsAsync(
instance.SiteId,
siteId => LoadDataConnectionsAsync(siteId, cancellationToken));
// Flatten
var flattenResult = _flatteningService.Flatten(
@@ -111,15 +137,20 @@ public class FlatteningPipeline : IFlatteningPipeline
var config = flattenResult.Value;
// Load shared scripts for semantic validation
var sharedScriptEntities = await _templateRepo.GetAllSharedScriptsAsync(cancellationToken);
var resolvedSharedScripts = sharedScriptEntities.Select(s => new ResolvedScript
// Load shared scripts for semantic validation. Session-global: identical
// for every instance in a batch, so hoisted to one query per session
// (the ArtifactDeploymentService global-hoist shape).
var resolvedSharedScripts = await session.GetSharedScriptsAsync(async () =>
{
CanonicalName = s.Name,
Code = s.Code,
ParameterDefinitions = s.ParameterDefinitions,
ReturnDefinition = s.ReturnDefinition
}).ToList();
var sharedScriptEntities = await _templateRepo.GetAllSharedScriptsAsync(cancellationToken);
return (IReadOnlyList<ResolvedScript>)sharedScriptEntities.Select(s => new ResolvedScript
{
CanonicalName = s.Name,
Code = s.Code,
ParameterDefinitions = s.ParameterDefinitions,
ReturnDefinition = s.ReturnDefinition
}).ToList();
});
// Compute the alarm-capable connection-name set so the semantic validator
// can gate native-alarm-source bindings. "Alarm-capable" matches the DCL
@@ -149,8 +180,13 @@ public class FlatteningPipeline : IFlatteningPipeline
// pre-loaded once into a name→JSON map here (avoiding sync-over-async) and the
// seam is a pure in-memory lookup. An unresolved {"$ref":"lib:Name"} in any
// validated script schema then becomes a deploy-blocking SchemaReference error.
var sharedSchemas = await _sharedSchemaRepo.ListAsync(cancellationToken);
var schemaLibrary = sharedSchemas.ToDictionary(s => s.Name, s => s.SchemaJson, StringComparer.Ordinal);
// Session-global, same as the shared scripts above.
var schemaLibrary = await session.GetSchemaLibraryAsync(async () =>
{
var sharedSchemas = await _sharedSchemaRepo.ListAsync(cancellationToken);
return (IReadOnlyDictionary<string, string>)sharedSchemas
.ToDictionary(s => s.Name, s => s.SchemaJson, StringComparer.Ordinal);
});
Func<string, string?> resolveSchemaRef = name => schemaLibrary.GetValueOrDefault(name);
// Validate. This is the deploy-gating path, so connection-binding completeness
@@ -180,24 +216,6 @@ public class FlatteningPipeline : IFlatteningPipeline
new FlatteningPipelineResult(config, hash, validation));
}
private async Task<IReadOnlyList<Commons.Entities.Templates.Template>> BuildTemplateChainAsync(
int templateId,
CancellationToken cancellationToken)
{
var chain = new List<Commons.Entities.Templates.Template>();
var currentId = (int?)templateId;
while (currentId.HasValue)
{
var template = await _templateRepo.GetTemplateWithChildrenAsync(currentId.Value, cancellationToken);
if (template == null) break;
chain.Add(template);
currentId = template.ParentTemplateId;
}
return chain;
}
private async Task<IReadOnlyDictionary<int, DataConnection>> LoadDataConnectionsAsync(
int siteId,
CancellationToken cancellationToken)
src/ZB.MOM.WW.ScadaBridge.DeploymentManager/IFlatteningPipeline.cs
+19 -1
View File
@@ -23,11 +23,29 @@ public interface IFlatteningPipeline
/// which need only the flattened config + revision hash, not a compile. Structural and
/// semantic validation still run in both cases.
/// </param>
/// <param name="session">
/// Optional flatten session memoising template-chain, composition and
/// session-global repository reads. Pass one shared session when flattening
/// several instances in a batch (see <c>DeploymentService.DeploySiteAsync</c>)
/// so a template chain shared by N instances is walked once. When
/// <see langword="null"/> the pipeline creates a private single-use session,
/// which still de-duplicates the repeated composed-chain loads inside one
/// instance's flatten. Additive parameter — existing callers are unaffected.
/// </param>
/// <returns>A task that resolves to the flattened configuration, revision hash, and validation result; or a failure result if flattening could not complete.</returns>
Task<Result<FlatteningPipelineResult>> FlattenAndValidateAsync(
int instanceId,
CancellationToken cancellationToken = default,
bool validateScripts = true);
bool validateScripts = true,
FlattenSession? session = null);
/// <summary>
/// Creates a fresh <see cref="FlattenSession"/> bound to this pipeline's
/// watermark, for a caller that wants to share one across a batch of
/// <see cref="FlattenAndValidateAsync"/> calls.
/// </summary>
/// <returns>A new, empty session.</returns>
FlattenSession CreateSession();
}
/// <summary>
src/ZB.MOM.WW.ScadaBridge.DeploymentManager/ServiceCollectionExtensions.cs
+8
View File
@@ -1,7 +1,9 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Templates;
namespace ZB.MOM.WW.ScadaBridge.DeploymentManager;
@@ -39,6 +41,12 @@ public static class ServiceCollectionExtensions
ServiceDescriptor.Singleton<IValidateOptions<DeploymentManagerOptions>, DeploymentManagerOptionsValidator>());
services.AddSingleton<OperationLockManager>();
// Template/instance version watermark backing the flatten-session cache key
// and the staleness fast path. TryAdd because AddConfigurationDatabase
// registers the same singleton — whichever composition root runs first wins
// and both observe one instance.
services.TryAddSingleton<ITemplateGraphWatermark, TemplateGraphWatermark>();
// Push-based deployment-status notification. Registered
// as a singleton so the scoped DeploymentService and the Central UI's
// scoped Blazor page component share one instance — both run in the
src/ZB.MOM.WW.ScadaBridge.DeploymentManager/StaleInstanceProbe.cs
+108 -3
View File
@@ -1,3 +1,5 @@
using System.Collections.Concurrent;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport;
namespace ZB.MOM.WW.ScadaBridge.DeploymentManager;
@@ -10,21 +12,70 @@ namespace ZB.MOM.WW.ScadaBridge.DeploymentManager;
/// deployed snapshot to decide staleness. Hosted in DeploymentManager so the
/// Transport component (which references only Commons + TemplateEngine) can probe
/// staleness without taking a DeploymentManager project reference.
///
/// <para>
/// <b>Watermark fast path.</b> A bundle import probes EVERY instance derived from
/// an overwritten template, and the Deployments page sweeps the fleet — both pay a
/// full flatten per instance for an answer that is usually "unchanged". A probe
/// result is therefore memoised against the
/// <see cref="ITemplateGraphWatermark"/> reading that produced it: the instance's
/// own version, the graph's structure version (which covers chain MEMBERSHIP), and
/// the version of every template the flatten actually walked. When all of those are
/// unchanged the recorded hash is still current by construction and the flatten is
/// skipped.
/// </para>
///
/// <para>
/// The fast path can only ever cause EXTRA work, never stale work: the watermark is
/// in-memory and process-local, so a restart, a failover to the other central node,
/// or any mutation at all simply misses and falls back to the authoritative
/// flatten.
/// </para>
/// </summary>
public sealed class StaleInstanceProbe : IStaleInstanceProbe
{
/// <summary>
/// Ceiling on memoised probe results. Bounded because the key space is the
/// instance table and this cache is process-wide and static; on overflow it is
/// dropped wholesale, which is safe here (unlike the script-compile verdict
/// cache) because a miss costs one flatten and leaks nothing.
/// </summary>
private const int MaxCachedProbes = 4096;
private static readonly ConcurrentDictionary<int, ProbeMemo> Memos = new();
private readonly IFlatteningPipeline _flatteningPipeline;
private readonly ITemplateGraphWatermark _watermark;
/// <summary>Initializes a new <see cref="StaleInstanceProbe"/>.</summary>
/// <param name="flatteningPipeline">The deployment flattening pipeline used to recompute the current revision hash.</param>
public StaleInstanceProbe(IFlatteningPipeline flatteningPipeline)
/// <param name="watermark">Graph version watermark backing the fast path.</param>
public StaleInstanceProbe(IFlatteningPipeline flatteningPipeline, ITemplateGraphWatermark watermark)
{
_flatteningPipeline = flatteningPipeline ?? throw new ArgumentNullException(nameof(flatteningPipeline));
_watermark = watermark ?? throw new ArgumentNullException(nameof(watermark));
}
/// <summary>
/// A previously computed revision hash together with the exact watermark
/// readings it was computed under.
/// </summary>
/// <param name="Structure">Graph structure version at flatten time.</param>
/// <param name="InstanceVersion">The instance's own version at flatten time.</param>
/// <param name="TemplateVersions">Version of every template the flatten walked, by template id.</param>
/// <param name="RevisionHash">The hash that reading produced.</param>
private sealed record ProbeMemo(
long Structure,
long InstanceVersion,
IReadOnlyDictionary<int, long> TemplateVersions,
string RevisionHash);
/// <inheritdoc />
public async Task<string?> GetCurrentRevisionHashAsync(int instanceId, CancellationToken cancellationToken = default)
{
if (TryGetMemoisedHash(instanceId, out var memoised))
return memoised;
// The pipeline returns a Result; a flatten failure (e.g. unresolvable
// template chain mid-import) yields null so the caller treats the
// instance as "staleness indeterminate" and skips it. Flattening reuses
@@ -33,9 +84,63 @@ public sealed class StaleInstanceProbe : IStaleInstanceProbe
// Staleness only needs the revision hash — skip the expensive Roslyn
// script-compilation stage (this probe is called per-instance across a whole
// bundle import in BundleImporter.ComputeStaleInstanceIdsAsync).
//
// Watermark readings are taken BEFORE the flatten. A mutation that lands
// during the flatten therefore bumps a version past the recorded one and
// invalidates the memo on the next read, rather than being captured as if
// it had already been included.
var structure = _watermark.StructureVersion;
var instanceVersion = _watermark.GetInstanceVersion(instanceId);
var session = _flatteningPipeline.CreateSession();
var result = await _flatteningPipeline
.FlattenAndValidateAsync(instanceId, cancellationToken, validateScripts: false)
.FlattenAndValidateAsync(instanceId, cancellationToken, validateScripts: false, session)
.ConfigureAwait(false);
return result.IsSuccess ? result.Value.RevisionHash : null;
if (result.IsFailure)
return null;
if (Memos.Count >= MaxCachedProbes)
Memos.Clear();
var visited = session.VisitedTemplateIds;
var versions = new Dictionary<int, long>(visited.Count);
foreach (var templateId in visited)
versions[templateId] = _watermark.GetTemplateVersion(templateId);
Memos[instanceId] = new ProbeMemo(structure, instanceVersion, versions, result.Value.RevisionHash);
return result.Value.RevisionHash;
}
/// <summary>
/// Returns the memoised revision hash when every watermark reading it was
/// computed under is still current.
/// </summary>
private bool TryGetMemoisedHash(int instanceId, out string? hash)
{
hash = null;
if (!Memos.TryGetValue(instanceId, out var memo))
return false;
if (memo.Structure != _watermark.StructureVersion)
return false;
if (memo.InstanceVersion != _watermark.GetInstanceVersion(instanceId))
return false;
foreach (var (templateId, version) in memo.TemplateVersions)
{
if (_watermark.GetTemplateVersion(templateId) != version)
return false;
}
hash = memo.RevisionHash;
return true;
}
/// <summary>
/// Drops every memoised probe result. Test hook — production invalidation is
/// driven entirely by the watermark.
/// </summary>
public static void ClearMemos() => Memos.Clear();
}