Files
ScadaBridge/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentService.cs
T
Joseph Doherty 9a87d7a328 Resolve all 33 medium-severity architecture review findings
Second slice of the 2026-09-09 whole-system review remediation
(docs/plans/2026-09-09-arch-review.md §10), on top of the High slice 146a1323.

Runtime/devices: SR-2 alarm on-trigger integration scope; SR-3 correlated NativeAlarmActor
subscribe retry; SR-4 immutable audited DB target; SR-5 ordered InstanceActor stop/flush;
SR-12 rehydration fence; SR-13 S&F gate recheck per delivery; DC-4 no OPC UA security
downgrade; DC-5 alarm snapshot for every new subscriber; ORCH-1 sync-dispose joins the pump.

Management/UI: MG-2 site scope on audit tree + Flow Debugger Replay; MG-3 stale-probe memo
removed; MG-4 server-owned bulk-deploy deadline; UI-5 sandbox runs in a killable child
process (new CentralUI.SandboxWorker project); UI-14 bounded sandbox output/IPC.

Edge/audit: EI-1 relay audit ownership; EI-2 SkipBodyCapture suppresses error text; EI-3/EI-11
error-field redaction + caps; EI-4 single relay deadline, pre-execution-only retry; EI-5
authenticated relay provenance; EI-10 body cap before audit read.

Persistence/contracts: DL-1 global Instance.UniqueName unique index + conflict translation;
DL-2 DataConnection protocol config encrypted at rest (migration
20260909203046_EnforceGlobalInstanceNamesAndProtectProtocolConfiguration — coordinated
central upgrade, old binaries cannot read protected rows); DL-3 bounded Replay paging;
DL-6 cross-site Overwrite of a deployed instance rejected; XC-4 proto descriptor contract test.

Config/hygiene: XC-1 central listener isolation by local port; XC-2/XC-3 inert SMTP + cron
keys removed from rig settings; CM-2 env2 bootstrap guard enabled; DC-6 deploy site exit code;
DC-7 notifier retries only explicit standby rejection; XC-5 ClusterClient purged from docs.

Gates recorded in the report: build 0/0, 9 397 passed / 0 failed, rig redeployed on all
eight nodes with the migration applied. 46 Low findings remain open.
2026-09-09 22:05:30 -04:00

2301 lines
112 KiB
C#

using System.Text.Json;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Lifecycle;
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
using ZB.MOM.WW.ScadaBridge.Communication;
using ZB.MOM.WW.ScadaBridge.TemplateEngine.Flattening;
using ZB.MOM.WW.ScadaBridge.TemplateEngine.Validation;
namespace ZB.MOM.WW.ScadaBridge.DeploymentManager;
/// <summary>
/// Central-side deployment orchestration service.
/// Coordinates the full deployment pipeline:
/// 1. Validate instance state transition
/// 2. Acquire per-instance operation lock
/// 3. Flatten configuration via TemplateEngine (captures template state at time of flatten)
/// 4. Validate flattened configuration
/// 5. Compute revision hash and diff
/// 6. Stage a PendingDeployment row + send RefreshDeploymentCommand (notify-and-fetch; site fetches the config over HTTP)
/// 7. Track deployment status with optimistic concurrency
/// 8. Store deployed config snapshot
/// 9. Audit log all actions
///
/// Each deployment has a unique deployment ID (GUID) + revision hash.
/// Template state captured at flatten time -- last-write-wins on templates is safe.
/// </summary>
public class DeploymentService
{
private readonly IDeploymentManagerRepository _repository;
private readonly ISiteRepository _siteRepository;
private readonly IFlatteningPipeline _flatteningPipeline;
private readonly CommunicationService _communicationService;
private readonly OperationLockManager _lockManager;
private readonly IAuditService _auditService;
private readonly DiffService _diffService;
private readonly RevisionHashService _revisionHashService;
private readonly IDeploymentStatusNotifier _statusNotifier;
private readonly IStaleInstanceProbe _staleInstanceProbe;
private readonly DeploymentManagerOptions _options;
private readonly CommunicationOptions _commOptions;
private readonly ILogger<DeploymentService> _logger;
/// <summary>
/// Prefix written to <see cref="DeploymentRecord.ErrorMessage"/> when a
/// deployment fails because the site command timed out or was cancelled.
/// Used by the query-before-redeploy trigger to tell
/// a timeout-induced failure apart from other deployment errors.
/// </summary>
private const string TimeoutFailurePrefix = "Communication failure:";
/// <summary>
/// Reason reported on every <see cref="SkippedInstanceResult"/> a bulk site
/// deploy produces for a <see cref="InstanceState.Disabled"/> instance.
/// </summary>
internal const string DisabledSkipReason =
"Disabled — bulk deploy preserves disabled state; deploy individually to re-enable";
/// <summary>
/// Reason reported on every <see cref="SkippedInstanceResult"/> a bulk site
/// deploy produces for an instance whose flattened config already matches the
/// revision the site is running. Suppressed by the <c>force</c> flag.
/// </summary>
internal const string UpToDateSkipReason =
"up-to-date — the flattened config still matches the deployed revision; " +
"redeploy with force to push it anyway";
/// <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>
/// <param name="repository">Repository for deployment manager data access.</param>
/// <param name="siteRepository">Repository for site data access.</param>
/// <param name="flatteningPipeline">Pipeline for flattening and validating template configurations.</param>
/// <param name="communicationService">Service for cross-cluster communication with sites.</param>
/// <param name="lockManager">Manager for per-instance operation locks.</param>
/// <param name="auditService">Service for recording audit log entries.</param>
/// <param name="diffService">Service for computing configuration diffs.</param>
/// <param name="revisionHashService">
/// Service for recomputing a flattened configuration's revision hash. Used by
/// <see cref="GetDeploymentComparisonAsync"/> to derive the deployed-side
/// staleness hash from the (List-normalized) deserialized snapshot — see I-1.
/// </param>
/// <param name="statusNotifier">Notifier for pushing deployment status changes to the UI.</param>
/// <param name="staleInstanceProbe">
/// Current-revision-hash probe backing the bulk deploy's up-to-date pre-filter
/// (see <see cref="DeploySiteAsync"/>). The same seam the bundle importer and
/// the Deployments page use, so all three agree on what "stale" means.
/// </param>
/// <param name="options">Deployment manager configuration options.</param>
/// <param name="communicationOptions">
/// Central-site communication options. The notify-and-fetch deploy path reads
/// <see cref="CommunicationOptions.CentralFetchBaseUrl"/> (carried in the
/// <c>RefreshDeploymentCommand</c> so sites need no standing config) and
/// <see cref="CommunicationOptions.PendingDeploymentTtl"/> (the staged
/// <c>PendingDeployment</c> row's lifetime).
/// </param>
/// <param name="logger">Logger instance.</param>
public DeploymentService(
IDeploymentManagerRepository repository,
ISiteRepository siteRepository,
IFlatteningPipeline flatteningPipeline,
CommunicationService communicationService,
OperationLockManager lockManager,
IAuditService auditService,
DiffService diffService,
RevisionHashService revisionHashService,
IDeploymentStatusNotifier statusNotifier,
IStaleInstanceProbe staleInstanceProbe,
IOptions<DeploymentManagerOptions> options,
IOptions<CommunicationOptions> communicationOptions,
ILogger<DeploymentService> logger)
{
_repository = repository;
_siteRepository = siteRepository;
_flatteningPipeline = flatteningPipeline;
_communicationService = communicationService;
_lockManager = lockManager;
_auditService = auditService;
_diffService = diffService;
_revisionHashService = revisionHashService;
_statusNotifier = statusNotifier;
_staleInstanceProbe = staleInstanceProbe;
_options = options.Value;
_commOptions = communicationOptions.Value;
_logger = logger;
}
/// <summary>
/// Raises a push notification that a deployment record's
/// status was just persisted, so the Central UI deployment-status page can
/// re-render over its SignalR circuit instead of polling. Called at every
/// point a <see cref="DeploymentRecord"/> status is written.
/// </summary>
private void NotifyStatusChange(DeploymentRecord record) =>
_statusNotifier.NotifyStatusChanged(
new DeploymentStatusChange(record.DeploymentId, record.InstanceId, record.Status));
/// <summary>
/// Resolves the site's string identifier from the numeric DB ID.
/// The communication layer routes by string identifier (e.g. "site-a"), not DB ID.
///
/// When the <see cref="Site"/> row is missing (FK was
/// deleted, race with admin delete, DB inconsistency) the previous behaviour
/// silently substituted the numeric id rendered as a string — every
/// downstream `CommunicationService` call then failed with a confusing
/// "unknown site" routing error that hid the real cause. Treat a missing
/// site row as a hard validation failure: throw
/// <see cref="InvalidOperationException"/> naming the unresolved id so the
/// operator sees the actual problem. On the deploy path the existing
/// try/catch turns this into a Failed deployment record with a clear
/// message; lifecycle paths propagate it to the caller (CLI/UI) which
/// surface it as an error to the operator.
/// </summary>
private async Task<string> ResolveSiteIdentifierAsync(int siteId, CancellationToken cancellationToken)
{
var site = await _siteRepository.GetSiteByIdAsync(siteId, cancellationToken);
if (site == null)
throw new InvalidOperationException(
$"Site with ID {siteId} not found; cannot resolve its SiteIdentifier for routing.");
return site.SiteIdentifier;
}
/// <summary>
/// Deploy an instance to its site.
/// Generates unique deployment ID, computes revision hash.
/// Validates state transitions, uses optimistic concurrency.
/// Site-side apply is all-or-nothing (handled by DeploymentManagerActor).
/// Stores deployed config snapshot on success.
/// Captures template state at time of flatten.
/// </summary>
/// <param name="instanceId">The database ID of the instance to deploy.</param>
/// <param name="user">The username initiating the deployment, recorded in the audit log.</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>A task that resolves to a success result containing the deployment record, or a failure result with an error message.</returns>
public async Task<Result<DeploymentRecord>> DeployInstanceAsync(
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.
// reconcileGate is null on this path on purpose: the batch-wide reconcile
// budget exists to stop a serial phase-1 loop multiplying one unreachable
// site's query timeout by the instance count. A single instance pays it
// once, and this is the explicit operator path — it keeps the full
// communication-layer QueryTimeout.
var prepared = await PrepareDeploymentAsync(
instanceId, user, session: null, reconcileGate: null, cancellationToken);
if (prepared.EarlyResult is { } early)
return early;
using (prepared.LockHandle)
{
var outcome = await StageAndSendDeploymentAsync(
prepared, user, dbGate: null, 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 and insert the
/// <c>InProgress</c> <see cref="DeploymentRecord"/>.
///
/// <para>
/// Note what is NOT here: the <c>PendingDeployment</c> row. Its fetch token
/// expires <see cref="CommunicationOptions.PendingDeploymentTtl"/> after it is
/// STAGED, so staging in phase 1 started the clock on every instance in a bulk
/// deploy at once while phase 2 reached them one batch at a time — with enough
/// instances the tail's tokens expired before their command was ever sent, and
/// the site's fetch 404'd. Staging therefore moved to
/// <see cref="StageAndSendDeploymentAsync"/>, immediately before each send.
/// </para>
///
/// <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="reconcileGate">
/// Batch-wide reconcile budget + reachability circuit-breaker for a bulk
/// deploy, or <see langword="null"/> on the single-instance path (which keeps
/// the communication layer's own query timeout).
/// </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,
BulkReconcileGate? reconcileGate,
CancellationToken cancellationToken)
{
// Load instance
var instance = await _repository.GetInstanceByIdAsync(instanceId, cancellationToken);
if (instance == null)
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 PreparedDeployment.Resolved(Result<DeploymentRecord>.Failure(transitionError));
// Acquire per-instance operation lock
var lockHandle = await _lockManager.AcquireAsync(
instance.UniqueName, _options.OperationLockTimeout, cancellationToken);
try
{
return await PrepareUnderLockAsync(
instance, user, session, reconcileGate, 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,
BulkReconcileGate? reconcileGate,
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, validateScripts: true, session);
if (flattenResult.IsFailure)
{
lockHandle.Dispose();
return PreparedDeployment.Resolved(
Result<DeploymentRecord>.Failure($"Validation failed: {flattenResult.Error}"));
}
var flattenedConfig = flattenResult.Value.Configuration;
var revisionHash = flattenResult.Value.RevisionHash;
var validationResult = flattenResult.Value.Validation;
if (!validationResult.IsValid)
{
// Return a grouped/summarized error (leading count + per-module
// rollup, capped) instead of a flat semicolon-joined dump that becomes a wall
// of text for instances with dozens of unbound attributes. The full per-entry
// list still goes to the deploy log for operators who need every clause.
_logger.LogWarning(
"Pre-deployment validation failed for instance {InstanceId} ({ErrorCount} error(s)): {Detail}",
instanceId,
validationResult.Errors.Count,
string.Join("; ", validationResult.Errors.Select(e => e.Message)));
lockHandle.Dispose();
return PreparedDeployment.Resolved(Result<DeploymentRecord>.Failure(
$"Pre-deployment validation failed: {validationResult.SummarizeErrors()}"));
}
// Serialize for transmission (also the payload stored in the deployed
// snapshot on success / reconciliation).
var configJson = JsonSerializer.Serialize(flattenedConfig);
// Query-the-site-before-redeploy idempotency.
// If a prior deployment for this instance is stuck InProgress or Failed
// due to a timeout, the site may have actually applied the config. Query
// the site for its currently-applied revision before re-sending so a
// duplicate deployment is not produced (design: "Deployment Identity &
// Idempotency"). A clean prior Success or a fresh first-time deploy
// skips this extra round-trip.
var reconciled = reconcileGate is null
? await TryReconcileWithSiteAsync(
instance, revisionHash, configJson, user, cancellationToken)
: await ReconcileUnderBatchGateAsync(
instance, revisionHash, configJson, user, reconcileGate, cancellationToken);
if (reconciled != null)
{
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
// here — BEFORE creating an InProgress record or staging a pending row —
// so the operator sees a clear configuration error instead of a
// confusing downstream site-fetch failure (and no InProgress record is
// stranded).
if (string.IsNullOrEmpty(_commOptions.CentralFetchBaseUrl))
{
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.
//
// The previous code wrote the record as Pending,
// then immediately updated it to InProgress with no work in between
// (flattening, validation, and reconciliation all completed above). The
// back-to-back write cost an extra SaveChangesAsync round-trip, an
// extra IDeploymentStatusNotifier push (which rendered a
// Pending→InProgress flicker for ~ms), and an extra row-version bump
// for nothing. The transient Pending slot carried no operational
// meaning — it was set and immediately overwritten — so dropping it
// collapses the start of the deploy into a single insert + notify.
// InProgress remains the documented "sent to site, awaiting response"
// state, set immediately before the round-trip below.
var record = new DeploymentRecord(deploymentId, user)
{
InstanceId = instanceId,
Status = DeploymentStatus.InProgress,
RevisionHash = revisionHash,
DeployedAt = DateTimeOffset.UtcNow
};
await _repository.AddDeploymentRecordAsync(record, cancellationToken);
await _repository.SaveChangesAsync(cancellationToken);
NotifyStatusChange(record);
try
{
// Site routing is resolved here (a repository read) so phase 2 needs
// nothing but the staging write itself.
var siteId = await ResolveSiteIdentifierAsync(instance.SiteId, cancellationToken);
return new PreparedDeployment(
instance, deploymentId, record, revisionHash, configJson,
siteId, lockHandle, EarlyResult: null);
}
catch (Exception ex)
{
// Preparation failed before anything was staged or 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>
/// Stages the deployment's <c>PendingDeployment</c> row and immediately sends
/// its <c>RefreshDeploymentCommand</c> — phase 2.
///
/// <para>
/// Notify-and-fetch: instead of shipping the (potentially oversized,
/// silently-dropped &gt;128 KB) flattened config inline in a
/// <c>DeployInstanceCommand</c>, the config is staged in a
/// <c>PendingDeployment</c> row and a small <c>RefreshDeploymentCommand</c> is
/// sent; the site then fetches the config from <c>CentralFetchBaseUrl</c> over
/// HTTP using the per-deployment fetch token.
/// </para>
///
/// <para>
/// <b>Staging is immediately before the send, on purpose.</b> The row's
/// <c>ExpiresAt</c> is <c>stagedAt + PendingDeploymentTtl</c> (5 min by
/// default) and the fetch endpoint enforces it, so the gap between staging and
/// sending is dead time burned off the token's life. Staging the whole batch up
/// front in phase 1 made that gap grow with batch size — for a large site the
/// last instances' tokens could expire before their command was even sent.
/// </para>
///
/// <para>
/// <b>Serialisation.</b> Staging is the ONE repository touch in this otherwise
/// network-only phase, and the scoped <c>DbContext</c> is not thread-safe, so a
/// bulk deploy passes <paramref name="dbGate"/> — a 1-permit semaphore that lets
/// exactly one in-flight send do its staging write at a time. The sends
/// themselves stay fully concurrent (they touch no repository), so this costs no
/// parallelism, and the single-instance path passes <see langword="null"/>
/// because there is nothing to serialise against.
/// </para>
///
/// <para>
/// A staging failure is returned as the phase-2 fault rather than thrown, so
/// phase 3 marks the record Failed on exactly the same path as a send fault —
/// the record must never be left InProgress.
/// </para>
/// </summary>
/// <param name="prepared">The prepared deployment to stage and send.</param>
/// <param name="user">User attributed on the refresh command.</param>
/// <param name="dbGate">Serialises the staging write across a concurrent batch; <see langword="null"/> for a batch of one.</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> StageAndSendDeploymentAsync(
PreparedDeployment prepared,
string user,
SemaphoreSlim? dbGate,
CancellationToken cancellationToken)
{
RefreshDeploymentCommand command;
try
{
if (dbGate is not null)
await dbGate.WaitAsync(cancellationToken);
try
{
command = await StagePendingDeploymentAsync(prepared, user, cancellationToken);
}
finally
{
dbGate?.Release();
}
}
catch (Exception ex)
{
return new SendOutcome(null, ex);
}
return await SendDeploymentAsync(prepared, command, cancellationToken);
}
/// <summary>
/// Writes the <c>PendingDeployment</c> row for a prepared deployment and builds
/// the matching <c>RefreshDeploymentCommand</c>, stamping <c>stagedAt</c> at the
/// moment of the write.
///
/// <para>
/// Cleanup of the staged row is TTL-based ONLY — it 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
/// to ≤1 per instance and the fetch endpoint enforces the TTL, so leaving rows
/// for TTL purge is safe. Expired rows are swept by the central
/// <c>PendingDeploymentPurgeActor</c> singleton on its
/// <see cref="CommunicationOptions.PendingDeploymentPurgeInterval"/> cadence.
/// </para>
/// </summary>
/// <param name="prepared">The prepared deployment being staged.</param>
/// <param name="user">User attributed on the refresh command.</param>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>The refresh command to send to the site.</returns>
private async Task<RefreshDeploymentCommand> StagePendingDeploymentAsync(
PreparedDeployment prepared,
string user,
CancellationToken cancellationToken)
{
var token = DeploymentFetchToken.Generate();
var stagedAt = DateTimeOffset.UtcNow;
await _repository.AddPendingDeploymentAsync(new PendingDeployment(
prepared.DeploymentId, prepared.Instance.Id, prepared.RevisionHash,
prepared.ConfigJson, token,
stagedAt, stagedAt + _commOptions.PendingDeploymentTtl), cancellationToken);
await _repository.SaveChangesAsync(cancellationToken);
return new RefreshDeploymentCommand(
prepared.DeploymentId, prepared.Instance.UniqueName, prepared.RevisionHash,
user, stagedAt, _commOptions.CentralFetchBaseUrl, token);
}
/// <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="command">The refresh command built by the staging step immediately before this call.</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,
RefreshDeploymentCommand command,
CancellationToken cancellationToken)
{
try
{
_logger.LogInformation(
"Sending deployment {DeploymentId} for instance {Instance} to site {SiteId} (notify-and-fetch)",
prepared.DeploymentId, prepared.Instance.UniqueName, prepared.SiteIdentifier);
cancellationToken.ThrowIfCancellationRequested();
var response = await _communicationService.RefreshDeploymentAsync(
prepared.SiteIdentifier, 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;
record.CompletedAt = DateTimeOffset.UtcNow;
// Once the site has confirmed the apply,
// commit the deployment record's terminal status BEFORE touching
// instance state and the deployed-config snapshot. If a later write
// (instance update / snapshot store) fails, the recorded fact that
// the site succeeded must NOT be lost -- otherwise central reports a
// non-Success record while the site is running the new config.
await _repository.UpdateDeploymentRecordAsync(record, cancellationToken);
await _repository.SaveChangesAsync(cancellationToken);
NotifyStatusChange(record);
if (response.Status == DeploymentStatus.Success)
{
// Telemetry: one instance deployment successfully applied to a
// site. Counted once per successful deploy operation (the unit
// of scadabridge.deployments.applied — one DeployInstanceAsync
// deploys exactly one instance to one site). Emitted only on this
// confirmed-Success path, so failures, timeouts/retries (the
// catch block), and the reconciliation path (which recovers a
// PRIOR timed-out apply rather than performing a fresh one) do
// not increment it.
ScadaBridgeTelemetry.RecordDeploymentApplied();
// The site has applied the deployment. The post-success
// persistence below is best-effort: a failure here must be
// logged loudly for operator reconciliation but must not flip
// the already-committed Success record back to Failed.
await ApplyPostSuccessSideEffectsAsync(
instance, deploymentId, prepared.RevisionHash, prepared.ConfigJson,
forceEnabledState: true, cancellationToken);
}
// Audit log. Best-effort and post-terminal: the deployment record's
// status is already committed above, so a failed audit write must NOT
// fall through to the outer catch and flip a committed Success back to
// Failed (audit is best-effort — the action's own result is authoritative).
await TryLogAuditAsync(user, "Deploy", "Instance", instanceId.ToString(),
instance.UniqueName, new { DeploymentId = deploymentId, Status = record.Status.ToString() },
cancellationToken);
_logger.LogInformation(
"Deployment {DeploymentId} for instance {Instance}: {Status}",
deploymentId, instance.UniqueName, record.Status);
return record.Status == DeploymentStatus.Success
? Result<DeploymentRecord>.Success(record)
: Result<DeploymentRecord>.Failure(
$"Deployment failed: {response.ErrorMessage ?? "Unknown error"}");
}
catch (Exception ex)
{
// 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 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>
///
/// <para>
/// <b>The cleanup save is un-poisoned before it is given up on (P7-5).</b> A
/// failed <c>SaveChanges</c> does not untrack the entity that caused it: EF
/// keeps it in the change set and re-attempts it on the NEXT save. Every phase
/// of a bulk site deploy shares ONE scoped <c>DbContext</c> across every
/// instance in the batch, so a persistent failure on one instance would
/// otherwise fail instance 2..N's status writes too — including their own
/// cleanup writes, which is exactly the write that stops a record being
/// stranded <c>InProgress</c>. On a cleanup-save failure the pending changes
/// are therefore DETACHED (<see cref="IDeploymentManagerRepository.DetachPendingChanges"/>)
/// and the Failed write is retried ONCE on the now-clean tracker: if the poison
/// came from an earlier instance this record still reaches its terminal status,
/// and either way the next instance in the batch saves cleanly.
/// </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 PersistFailedStatusAsync();
}
catch (Exception cleanupEx)
{
// Give up on whatever is pending — it has already failed to save at
// least once — so the shared change tracker stops re-attempting it.
var detached = _repository.DetachPendingChanges();
_logger.LogError(cleanupEx,
"Failed to persist Failed status for deployment {DeploymentId} of instance {Instance} " +
"after deployment error: {Error}. Detached {DetachedCount} pending change(s) so the " +
"shared unit of work does not carry the failure into later writes",
deploymentId, instance.UniqueName, cause.Message, detached);
if (detached > 0)
{
try
{
await PersistFailedStatusAsync();
}
catch (Exception retryEx)
{
_logger.LogError(retryEx,
"Retrying the Failed status write for deployment {DeploymentId} of instance " +
"{Instance} on a cleaned unit of work also failed; the record remains " +
"InProgress and must be reconciled (the next deploy of this instance heals it)",
deploymentId, instance.UniqueName);
}
}
}
// Best-effort and deliberately OUTSIDE the status-write try: an audit row
// for a failed deployment is worth writing even when the status write
// itself could not be persisted, and an audit failure must never mask the
// deployment error the caller is about to report.
await TryLogAuditAsync(user, "DeployFailed", "Instance", instance.Id.ToString(),
instance.UniqueName, new { DeploymentId = deploymentId, Error = cause.Message },
CancellationToken.None);
// Writes the terminal Failed status and pushes the status change.
//
// 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.
async Task PersistFailedStatusAsync()
{
await _repository.UpdateDeploymentRecordAsync(record, CancellationToken.None);
await _repository.SaveChangesAsync(CancellationToken.None);
NotifyStatusChange(record);
}
}
/// <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,
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, 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>
/// A bulk deploy's shared query-before-redeploy budget: the short per-reconcile
/// deadline every instance in the batch runs under, plus the once-per-batch
/// reachability circuit-breaker that skips the rest after the first timeout.
/// See <see cref="ReconcileUnderBatchGateAsync"/> for why both bounds exist.
///
/// <para>
/// Deliberately not thread-safe: it is only ever touched from the serial
/// prepare phase.
/// </para>
/// </summary>
/// <param name="reconcileTimeout">Deadline applied to each reconcile round-trip.</param>
private sealed class BulkReconcileGate(TimeSpan reconcileTimeout)
{
/// <summary>Deadline applied to each reconcile round-trip in this batch.</summary>
public TimeSpan ReconcileTimeout { get; } = reconcileTimeout;
/// <summary>True once a reconcile in this batch hit its deadline.</summary>
public bool SiteUnreachable { get; private set; }
/// <summary>Number of reconciles the circuit-breaker skipped, for the batch's one summary log line.</summary>
public int SkippedCount { get; private set; }
/// <summary>Opens the circuit for the remainder of the batch.</summary>
public void MarkUnreachable() => SiteUnreachable = true;
/// <summary>Counts one reconcile skipped because the circuit is open.</summary>
public void CountSkipped() => SkippedCount++;
}
/// <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 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>Stage + 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. Each instance's
/// <c>PendingDeployment</c> row is staged here, immediately before its own
/// send, so a fetch token's TTL is not burned waiting for the batch ahead of
/// it; that single repository touch is serialised behind a 1-permit
/// semaphore, since the <c>DbContext</c> is not thread-safe.
/// </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 — with two batch-level differences. 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>
///
/// <para>
/// <b>Disabled instances are skipped entirely</b> and reported in
/// <see cref="SiteDeploymentSummary.SkippedInstances"/> — see the filter below
/// for why. <b>Up-to-date instances are skipped too</b> unless
/// <paramref name="force"/> is set, for the reason documented on
/// <see cref="IsUpToDateAsync"/>: a redeploy is destructive to site-local
/// state, so pushing one to an instance whose config has not changed loses
/// data for no benefit. <b>Reconciliation is bounded per batch</b> by
/// <see cref="ReconcileUnderBatchGateAsync"/>, so an unreachable site costs the
/// batch one short timeout rather than one 30-second query per instance.
/// </para>
///
/// <para>
/// The two filters compose in one direction only: <b>Disabled wins</b>. A
/// Disabled instance is never deployed at bulk scope, force or not — force
/// suppresses the up-to-date filter, never the Disabled one.
/// </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="force">
/// Deploy every deployable instance even when its config is unchanged — the
/// pre-2026-08 behaviour. The operator escape hatch for the case central's
/// recorded state cannot see: a site that lost its config (restored volume,
/// rebuilt node) looks up-to-date from here and needs the push anyway. Has no
/// effect on Disabled instances.
/// </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,
bool force = false,
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, []));
// Deliberately-Disabled instances are NOT part of a bulk deploy. Deploy is a
// legal transition out of Disabled and the single-instance path uses it to
// deploy-and-enable, so including them here would have one site-wide push
// silently START every piece of equipment an operator had deliberately
// stopped. They are reported as skipped rows instead — no flatten, no send,
// no state change — and remain individually deployable, which is the
// explicit "yes, re-enable this one" action.
//
// Unchanged instances are then filtered out for the mirror-image reason:
// see IsUpToDateAsync.
var deployable = new List<Instance>(instances.Count);
var skipped = new List<SkippedInstanceResult>();
var disabledCount = 0;
foreach (var instance in instances)
{
cancellationToken.ThrowIfCancellationRequested();
if (instance.State == InstanceState.Disabled)
{
skipped.Add(new SkippedInstanceResult(instance.Id, instance.UniqueName, DisabledSkipReason));
disabledCount++;
}
else if (!force && await IsUpToDateAsync(instance, cancellationToken))
{
skipped.Add(new SkippedInstanceResult(instance.Id, instance.UniqueName, UpToDateSkipReason));
}
else
{
deployable.Add(instance);
}
}
var upToDateCount = skipped.Count - disabledCount;
// ---- Phase 1: prepare, serially, on ONE shared flatten session. ----
var session = _flatteningPipeline.CreateSession();
var reconcileGate = new BulkReconcileGate(_options.SiteDeploymentReconcileTimeout);
var prepared = new List<PreparedDeployment>(deployable.Count);
var results = new List<InstanceDeploymentResult>(deployable.Count);
// Index of the first prepared deployment phase 3 has NOT yet finalised.
// Everything from here to the end of `prepared` still owns an operation
// lock and an InProgress record, so the cleanup handler below knows exactly
// what it must unwind if the operation escapes early.
var finalizedUpTo = 0;
try
{
foreach (var instance in deployable)
{
cancellationToken.ThrowIfCancellationRequested();
PreparedDeployment step;
try
{
step = await PrepareDeploymentAsync(
instance.Id, user, session, reconcileGate, 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: stage + bounded-parallel site round-trips. ----
// Never throws: faults (cancellation included) come back as per-instance
// outcomes so phase 3 still runs for every prepared deployment.
var outcomes = await SendPreparedAsync(prepared, user, cancellationToken);
// ---- Phase 3: finalize, serially, releasing each lock as we go. ----
for (; finalizedUpTo < prepared.Count; finalizedUpTo++)
{
var step = prepared[finalizedUpTo];
try
{
var result = await FinalizeDeploymentAsync(step, outcomes[finalizedUpTo], user, cancellationToken);
results.Add(ToInstanceResult(step.Instance.Id, step.Instance.UniqueName, result, step.DeploymentId));
}
finally
{
step.LockHandle?.Dispose();
}
}
}
catch (Exception ex)
{
// Nothing above may escape while a prepared deployment still holds its
// per-instance operation lock: OperationLockManager hands out real
// semaphores, so an undisposed handle wedges that instance against
// EVERY future mutating command for the life of the process — a
// permanent, restart-only outage for one instance. The two live escape
// routes are phase 1's ThrowIfCancellationRequested and a fault from
// phase 3's own persistence; both land here.
//
// Records are finalised as Failed for the same reason the send/persist
// paths do it: a deployment must never be left InProgress. Uses
// CancellationToken.None inside (see MarkDeploymentFailedAsync) so the
// already-cancelled token cannot abort the cleanup writes themselves.
await ReleasePreparedAsync(prepared, finalizedUpTo, user, ex);
if (ex is not OperationCanceledException || !cancellationToken.IsCancellationRequested) throw;
var recorded = results.Select(r => r.InstanceId).ToHashSet();
foreach (var pending in deployable.Where(i => !recorded.Contains(i.Id)))
results.Add(new InstanceDeploymentResult(pending.Id, pending.UniqueName, null, false,
"Overall bulk deployment deadline exceeded; no further apply was scheduled."));
}
var summary = new SiteDeploymentSummary(
site.SiteIdentifier,
results,
results.Count(r => r.Success),
results.Count(r => !r.Success),
skipped);
// ONE summary line for the whole batch, never one per skipped reconcile:
// an unreachable site would otherwise log N warnings for a single fact.
if (reconcileGate.SkippedCount > 0)
{
_logger.LogWarning(
"Site {SiteId} did not answer the query-before-redeploy reconciliation within {Timeout}; " +
"skipped it for the remaining {SkippedCount} instance(s) of this bulk deploy — they proceed " +
"on central's recorded state, with the site's own stale-revision rejection as the safety net",
site.SiteIdentifier, reconcileGate.ReconcileTimeout, reconcileGate.SkippedCount);
}
_logger.LogInformation(
"Bulk deployment of site {SiteId} complete: {SuccessCount} succeeded, {FailureCount} failed, " +
"{DisabledCount} skipped as Disabled, {UpToDateCount} skipped as up-to-date (force={Force}) " +
"({ChainLoads} template-chain load(s) across {InstanceCount} instance(s))",
site.SiteIdentifier, summary.SuccessCount, summary.FailureCount, disabledCount, upToDateCount,
force, session.ChainLoads, deployable.Count);
return Result<SiteDeploymentSummary>.Success(summary);
}
/// <summary>
/// True when a bulk site deploy should leave this instance alone because the
/// site is already running the config it would be sent.
///
/// <para>
/// <b>Why an unconditional bulk redeploy is the wrong default.</b> A redeploy
/// is NOT idempotent at the site: applying a config resets that instance's
/// static attribute overrides and clears its native alarm state, both by
/// design. A site-wide push therefore destroyed live operational state on every
/// instance it touched, including the ones whose configuration had not changed
/// at all — the usual case, since the operator is normally pushing one edited
/// template. Skipping them costs nothing (the site is already running that
/// revision) and preserves the state.
/// </para>
///
/// <para>
/// <b>The test is deliberately conservative.</b> Every uncertainty resolves to
/// "deploy it", so this filter can only ever fail to skip — it can never skip
/// an instance that needed the push:
/// </para>
/// <list type="bullet">
/// <item>
/// Only an <see cref="InstanceState.Enabled"/> instance can be up to date.
/// A <see cref="InstanceState.NotDeployed"/> one is not running the config
/// no matter what a leftover snapshot row says.
/// </item>
/// <item>
/// No <see cref="DeployedConfigSnapshot"/> means central has never seen a
/// successful apply — nothing to compare against.
/// </item>
/// <item>
/// A prior record that <see cref="ShouldQuerySiteBeforeRedeploy"/> flags
/// (stuck <c>InProgress</c>, or timeout-<c>Failed</c>) must go through
/// prepare, because that is where the query-before-redeploy reconciliation
/// that HEALS such a record runs. Skipping here would strand it forever.
/// </item>
/// <item>
/// The probe returning <see langword="null"/> (unresolvable template chain)
/// or throwing means staleness is indeterminate.
/// </item>
/// </list>
///
/// <para>
/// The comparison uses the snapshot's own stored <c>RevisionHash</c> rather
/// than <see cref="GetDeploymentComparisonAsync"/>'s recomputed-from-normalized
/// hash: the normalization exists to suppress a SPURIOUS stale flag (I-1), and
/// treating a spuriously-stale instance as stale here just deploys it, which is
/// the old behaviour and always safe.
/// </para>
///
/// <para>
/// <b>Cost.</b> The probe re-flattens on its own session rather than the
/// batch's, so a fully-stale site pays one extra flatten per instance before
/// prepare re-flattens it for real. Each request checks authoritative database
/// state because a peer central node can mutate configuration without bumping
/// this process's watermark. Unchanged instances still avoid site round-trips.
/// </para>
/// </summary>
/// <param name="instance">The candidate instance.</param>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns><see langword="true"/> only when the instance is provably unchanged.</returns>
private async Task<bool> IsUpToDateAsync(Instance instance, CancellationToken cancellationToken)
{
if (instance.State != InstanceState.Enabled)
return false;
var snapshot = await _repository.GetDeployedSnapshotByInstanceIdAsync(instance.Id, cancellationToken);
if (snapshot == null)
return false;
var prior = await _repository.GetCurrentDeploymentStatusAsync(instance.Id, cancellationToken);
if (prior != null && ShouldQuerySiteBeforeRedeploy(prior))
return false;
string? currentHash;
try
{
currentHash = await _staleInstanceProbe.GetCurrentRevisionHashAsync(instance.Id, cancellationToken);
}
catch (Exception ex) when (ex is not OperationCanceledException)
{
// Staleness is an optimisation, never a gate: an unexpected probe fault
// falls back to deploying the instance, which is what this method's
// absence would have done.
_logger.LogWarning(ex,
"Could not determine whether instance {Instance} is up to date; deploying it",
instance.UniqueName);
return false;
}
return currentHash != null
&& string.Equals(currentHash, snapshot.RevisionHash, StringComparison.Ordinal);
}
/// <summary>
/// Unwinds the prepared deployments a bulk deploy never finalised, after the
/// operation escaped early (cancellation, or a fault in phase 3's own
/// persistence). For each one it writes the terminal Failed status and releases
/// the per-instance operation lock.
///
/// <para>
/// Best-effort throughout: a fault while unwinding one entry must not stop the
/// remaining locks from being released, and must not mask the original
/// exception the caller is about to rethrow. The lock release in particular is
/// in a <c>finally</c>, because leaking it is the unrecoverable outcome — the
/// Failed status can be reconciled by an operator, a wedged semaphore cannot.
/// </para>
/// </summary>
/// <param name="prepared">The batch's prepared deployments.</param>
/// <param name="from">Index of the first entry phase 3 did not finalise.</param>
/// <param name="user">User attributed on the failure audit rows.</param>
/// <param name="cause">The exception that aborted the operation.</param>
/// <returns>A task that completes once every outstanding entry has been unwound.</returns>
private async Task ReleasePreparedAsync(
IReadOnlyList<PreparedDeployment> prepared,
int from,
string user,
Exception cause)
{
for (var i = from; i < prepared.Count; i++)
{
var step = prepared[i];
try
{
await MarkDeploymentFailedAsync(
step.Record, step.Instance, step.DeploymentId, user, cause);
}
catch (Exception ex)
{
_logger.LogError(ex,
"Failed to finalize deployment {DeploymentId} for instance {Instance} while unwinding " +
"an aborted bulk deployment",
step.DeploymentId, step.Instance.UniqueName);
}
finally
{
step.LockHandle?.Dispose();
}
}
}
/// <summary>
/// Runs phase 2 for a whole batch: every prepared deployment's staging write
/// plus 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.
///
/// <para>
/// Never throws on cancellation: the whole point of returning outcomes
/// positionally is that phase 3 must run for EVERY prepared deployment, whatever
/// happened in phase 2 — otherwise records stay InProgress and their operation
/// locks stay held. A cancelled instance is recorded as its own
/// <see cref="SendOutcome"/> fault and finalised as Failed like any other.
/// </para>
/// </summary>
/// <param name="prepared">The prepared deployments to stage and send.</param>
/// <param name="user">User attributed on each refresh command.</param>
/// <param name="cancellationToken">Cancellation token for the batch.</param>
/// <returns>One outcome per prepared deployment, positionally aligned.</returns>
private async Task<SendOutcome[]> SendPreparedAsync(
IReadOnlyList<PreparedDeployment> prepared,
string user,
CancellationToken cancellationToken)
{
var outcomes = new SendOutcome[prepared.Count];
if (prepared.Count == 0)
return outcomes;
using var gate = new SemaphoreSlim(_options.SiteDeploymentMaxParallelism);
// Serialises the ONE repository touch left in this phase (the staging
// write) across the concurrent sends — the scoped DbContext is not
// thread-safe. The sends themselves stay fully concurrent.
using var dbGate = new SemaphoreSlim(1, 1);
var sends = prepared.Select(async (step, index) =>
{
// Cancellation must not escape as an exception: it would abandon phase 3
// for every instance, leaking their operation locks (a wedged
// per-instance semaphore is permanent for the life of the process) and
// stranding their records InProgress.
try
{
await gate.WaitAsync(cancellationToken);
}
catch (OperationCanceledException ex)
{
outcomes[index] = new SendOutcome(null, ex);
return;
}
try
{
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cts.CancelAfter(_options.SiteDeploymentTimeoutPerInstance);
outcomes[index] = await StageAndSendDeploymentAsync(step, user, dbGate, cts.Token);
}
catch (Exception ex)
{
// StageAndSendDeploymentAsync already converts its own faults into
// outcomes; this is the belt-and-braces net so no send task can
// fault Task.WhenAll and skip phase 3.
outcomes[index] = new SendOutcome(null, ex);
}
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>
/// <param name="instanceId">The database ID of the instance to disable.</param>
/// <param name="user">The username initiating the operation, recorded in the audit log.</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>A task that resolves to a success result with the site response, or a failure result with an error message.</returns>
public async Task<Result<InstanceLifecycleResponse>> DisableInstanceAsync(
int instanceId,
string user,
CancellationToken cancellationToken = default)
{
var instance = await _repository.GetInstanceByIdAsync(instanceId, cancellationToken);
if (instance == null)
return Result<InstanceLifecycleResponse>.Failure($"Instance with ID {instanceId} not found.");
var transitionError = StateTransitionValidator.ValidateTransition(instance.State, "disable");
if (transitionError != null)
return Result<InstanceLifecycleResponse>.Failure(transitionError);
using var lockHandle = await _lockManager.AcquireAsync(
instance.UniqueName, _options.OperationLockTimeout, cancellationToken);
var commandId = Guid.NewGuid().ToString("N");
var siteId = await ResolveSiteIdentifierAsync(instance.SiteId, cancellationToken);
var command = new DisableInstanceCommand(commandId, instance.UniqueName, DateTimeOffset.UtcNow);
// Bound the round-trip with the configured lifecycle timeout so a
// hung/unreachable site does not block the operation lock indefinitely.
InstanceLifecycleResponse response;
try
{
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cts.CancelAfter(_options.LifecycleCommandTimeout);
response = await _communicationService.DisableInstanceAsync(siteId, command, cts.Token);
}
catch (Exception ex) when (ex is TimeoutException or OperationCanceledException)
{
// A lifecycle command timeout produced no
// audit row pre-fix — the operator saw a timeout in the UI but
// the audit trail showed nothing happened, contrary to the
// design's "audit logging for all instance lifecycle changes"
// rule. Mirror the DeployFailed pattern: write a "<Action>TimedOut"
// entry with CancellationToken.None so a cancelled outer token
// (the typical reason this catch ran) cannot prevent the
// durable audit write.
await TryLogLifecycleTimeoutAsync(
user, "DisableTimedOut", instanceId, instance.UniqueName, commandId, ex);
_logger.LogWarning(ex, "Disable of instance {Instance} timed out", instance.UniqueName);
return Result<InstanceLifecycleResponse>.Failure(
$"Disable failed: the site did not respond within {_options.LifecycleCommandTimeout}.");
}
if (response.Success)
{
instance.State = InstanceState.Disabled;
await _repository.UpdateInstanceAsync(instance, cancellationToken);
await _repository.SaveChangesAsync(cancellationToken);
}
await TryLogAuditAsync(user, "Disable", "Instance", instanceId.ToString(),
instance.UniqueName, new { CommandId = commandId, response.Success },
cancellationToken);
return response.Success
? Result<InstanceLifecycleResponse>.Success(response)
: Result<InstanceLifecycleResponse>.Failure(response.ErrorMessage ?? "Disable failed.");
}
/// <summary>
/// Enable an instance. Re-creates Instance Actor from stored config.
/// </summary>
/// <param name="instanceId">The database ID of the instance to enable.</param>
/// <param name="user">The username initiating the operation, recorded in the audit log.</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>A task that resolves to a success result with the site response, or a failure result with an error message.</returns>
public async Task<Result<InstanceLifecycleResponse>> EnableInstanceAsync(
int instanceId,
string user,
CancellationToken cancellationToken = default)
{
var instance = await _repository.GetInstanceByIdAsync(instanceId, cancellationToken);
if (instance == null)
return Result<InstanceLifecycleResponse>.Failure($"Instance with ID {instanceId} not found.");
var transitionError = StateTransitionValidator.ValidateTransition(instance.State, "enable");
if (transitionError != null)
return Result<InstanceLifecycleResponse>.Failure(transitionError);
using var lockHandle = await _lockManager.AcquireAsync(
instance.UniqueName, _options.OperationLockTimeout, cancellationToken);
var commandId = Guid.NewGuid().ToString("N");
var siteId = await ResolveSiteIdentifierAsync(instance.SiteId, cancellationToken);
var command = new EnableInstanceCommand(commandId, instance.UniqueName, DateTimeOffset.UtcNow);
// Bound the round-trip with the configured lifecycle timeout.
InstanceLifecycleResponse response;
try
{
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cts.CancelAfter(_options.LifecycleCommandTimeout);
response = await _communicationService.EnableInstanceAsync(siteId, command, cts.Token);
}
catch (Exception ex) when (ex is TimeoutException or OperationCanceledException)
{
// Emit an audit entry on lifecycle timeout
// so the operator's attempted Enable is recorded; see the matching
// comment in DisableInstanceAsync for the full rationale.
await TryLogLifecycleTimeoutAsync(
user, "EnableTimedOut", instanceId, instance.UniqueName, commandId, ex);
_logger.LogWarning(ex, "Enable of instance {Instance} timed out", instance.UniqueName);
return Result<InstanceLifecycleResponse>.Failure(
$"Enable failed: the site did not respond within {_options.LifecycleCommandTimeout}.");
}
if (response.Success)
{
instance.State = InstanceState.Enabled;
await _repository.UpdateInstanceAsync(instance, cancellationToken);
await _repository.SaveChangesAsync(cancellationToken);
}
await TryLogAuditAsync(user, "Enable", "Instance", instanceId.ToString(),
instance.UniqueName, new { CommandId = commandId, response.Success },
cancellationToken);
return response.Success
? Result<InstanceLifecycleResponse>.Success(response)
: Result<InstanceLifecycleResponse>.Failure(response.ErrorMessage ?? "Enable failed.");
}
/// <summary>
/// Delete an instance. Stops the site actor, removes site config, and
/// removes the central instance record (deployment history, snapshot,
/// overrides, and connection bindings go with it). S&amp;F NOT cleared.
/// Delete fails if the site is unreachable within
/// <c>CommunicationOptions.LifecycleTimeout</c> (applied inside
/// <see cref="CommunicationService.DeleteInstanceAsync"/>).
/// </summary>
/// <param name="instanceId">The database ID of the instance to delete.</param>
/// <param name="user">The username initiating the deletion, recorded in the audit log.</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>A task that resolves to a success result with the site response, or a failure result with an error message.</returns>
public async Task<Result<InstanceLifecycleResponse>> DeleteInstanceAsync(
int instanceId,
string user,
CancellationToken cancellationToken = default)
{
var instance = await _repository.GetInstanceByIdAsync(instanceId, cancellationToken);
if (instance == null)
return Result<InstanceLifecycleResponse>.Failure($"Instance with ID {instanceId} not found.");
var transitionError = StateTransitionValidator.ValidateTransition(instance.State, "delete");
if (transitionError != null)
return Result<InstanceLifecycleResponse>.Failure(transitionError);
using var lockHandle = await _lockManager.AcquireAsync(
instance.UniqueName, _options.OperationLockTimeout, cancellationToken);
var commandId = Guid.NewGuid().ToString("N");
// A NotDeployed instance has no live Instance Actor at any site, so Delete is
// a pure central-side record cleanup -- there is nothing to tear down at the
// site and no reason to require the site be reachable. Transport-imported
// instances always land NotDeployed (often against an uncommissioned or
// unreachable site), so a site round-trip here would make them undeletable
// until their site comes online. Skip the site entirely and remove the record.
if (instance.State == InstanceState.NotDeployed)
{
try
{
await _repository.DeleteInstanceAsync(instanceId, cancellationToken);
await _repository.SaveChangesAsync(cancellationToken);
}
catch (Exception ex)
{
_logger.LogError(ex,
"Failed to delete NotDeployed instance {Instance} centrally", instance.UniqueName);
return Result<InstanceLifecycleResponse>.Failure(
$"Delete failed: the central record for '{instance.UniqueName}' could not be removed: {ex.Message}.");
}
await TryLogAuditAsync(user, "Delete", "Instance", instanceId.ToString(),
instance.UniqueName,
new { CommandId = commandId, Success = true, SiteRoundTripSkipped = true },
cancellationToken);
return Result<InstanceLifecycleResponse>.Success(new InstanceLifecycleResponse(
commandId, instance.UniqueName, Success: true, ErrorMessage: null, DateTimeOffset.UtcNow));
}
var siteId = await ResolveSiteIdentifierAsync(instance.SiteId, cancellationToken);
var command = new DeleteInstanceCommand(commandId, instance.UniqueName, DateTimeOffset.UtcNow);
// Bound the round-trip with the configured lifecycle timeout.
InstanceLifecycleResponse response;
try
{
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cts.CancelAfter(_options.LifecycleCommandTimeout);
response = await _communicationService.DeleteInstanceAsync(siteId, command, cts.Token);
}
catch (Exception ex) when (ex is TimeoutException or OperationCanceledException)
{
// Emit an audit entry on lifecycle timeout
// so the operator's attempted Delete is recorded; see the matching
// comment in DisableInstanceAsync for the full rationale.
await TryLogLifecycleTimeoutAsync(
user, "DeleteTimedOut", instanceId, instance.UniqueName, commandId, ex);
_logger.LogWarning(ex, "Delete of instance {Instance} timed out", instance.UniqueName);
return Result<InstanceLifecycleResponse>.Failure(
$"Delete failed: the site did not respond within {_options.LifecycleCommandTimeout}.");
}
if (response.Success)
{
// Delete means delete: remove the instance record entirely.
// Deployment records, snapshot, overrides, and connection bindings
// are removed with it (see repository implementation).
//
// The site has already destroyed the Instance
// Actor and removed its config. If the central record removal now
// fails (DB error / concurrency), the exception must NOT escape
// uncaught -- that would leave the central record orphaned and
// un-deletable through the normal path (a re-issued delete may fail
// because the site no longer has the instance). Surface a distinct
// failure so an operator can reconcile.
try
{
await _repository.DeleteInstanceAsync(instanceId, cancellationToken);
await _repository.SaveChangesAsync(cancellationToken);
}
catch (Exception ex)
{
_logger.LogError(ex,
"Instance {Instance} was deleted at the site, but the central record could not be " +
"removed -- the central record is now orphaned and must be reconciled manually",
instance.UniqueName);
await TryLogAuditAsync(user, "DeleteOrphaned", "Instance", instanceId.ToString(),
instance.UniqueName, new { CommandId = commandId, Error = ex.Message },
CancellationToken.None);
return Result<InstanceLifecycleResponse>.Failure(
$"The site deleted instance '{instance.UniqueName}', but the central record could not " +
$"be removed: {ex.Message}. The central record is orphaned and must be reconciled.");
}
}
await TryLogAuditAsync(user, "Delete", "Instance", instanceId.ToString(),
instance.UniqueName, new { CommandId = commandId, response.Success },
cancellationToken);
return response.Success
? Result<InstanceLifecycleResponse>.Success(response)
: Result<InstanceLifecycleResponse>.Failure(
response.ErrorMessage ?? "Delete failed. Site may be unreachable.");
}
/// <summary>
/// Get the deployed config snapshot and compare with current
/// template-derived state. Produces both a staleness flag and — per the
/// design's "Diff View" — a structured <see cref="ConfigurationDiff"/> of
/// added/removed/changed attributes, alarms, and scripts (including data
/// connection binding changes) computed by the TemplateEngine
/// <see cref="DiffService"/>.
/// </summary>
/// <param name="instanceId">The database ID of the instance to compare.</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>A task that resolves to a success result with the comparison, or a failure result if no snapshot exists.</returns>
public async Task<Result<DeploymentComparisonResult>> GetDeploymentComparisonAsync(
int instanceId,
CancellationToken cancellationToken = default)
{
var snapshot = await _repository.GetDeployedSnapshotByInstanceIdAsync(instanceId, cancellationToken);
if (snapshot == null)
return Result<DeploymentComparisonResult>.Failure("No deployed snapshot found for this instance.");
// Compute current template-derived config. This read-only comparison only needs
// the revision hash (staleness) + the flattened config (structured diff) — NOT a
// script compile — so skip the expensive Roslyn ValidateScriptCompilation stage.
// The deploy gate remains the authoritative compile.
var currentResult = await _flatteningPipeline.FlattenAndValidateAsync(
instanceId, cancellationToken, validateScripts: false);
if (currentResult.IsFailure)
return Result<DeploymentComparisonResult>.Failure($"Cannot compute current config: {currentResult.Error}");
var currentConfig = currentResult.Value.Configuration;
var currentHash = currentResult.Value.RevisionHash;
// I-1 (latent): the snapshot's ConfigurationJson + RevisionHash froze the
// FLATTENED config at deploy time. The current config is a FRESH flatten,
// now always in native List form (consolidated element-type/coercion
// into AttributeValueCodec, which emits native-form JSON arrays). A List
// attribute deployed in the OLD quoted form (e.g. ["10","20"]) therefore
// both (a) hashes differently from the native re-flatten — a spurious
// stale flag — and (b) shows a spurious Changed attribute in the diff
// (DiffService.AttributesEqual is an ordinal Value comparison). Normalize
// the deserialized snapshot's List values through AttributeValueCodec
// Decode→Encode so an old-form value becomes native form and compares
// equal to the native re-flatten, then drive BOTH the staleness hash and
// the diff off that normalized snapshot. Scalars are left untouched.
//
// A snapshot that cannot be deserialized (corrupt /
// older schema) still yields the frozen-hash staleness result, with a
// null diff.
var deployedRevisionHash = snapshot.RevisionHash;
ConfigurationDiff? diff = null;
try
{
var deployedConfig = JsonSerializer.Deserialize<FlattenedConfiguration>(snapshot.ConfigurationJson);
if (deployedConfig != null)
{
deployedConfig = NormalizeListAttributeValues(deployedConfig);
// Recompute the deployed-side hash from the normalized snapshot so
// an old-form List value is not flagged stale against the native
// re-flatten. For a faithfully-stored scalar-only snapshot this
// reproduces the frozen RevisionHash exactly, so behaviour is
// unchanged outside the List-normalization case.
deployedRevisionHash = _revisionHashService.ComputeHash(deployedConfig);
diff = _diffService.ComputeDiff(
deployedConfig, currentConfig, deployedRevisionHash, currentHash);
}
else
{
_logger.LogWarning(
"Deployed snapshot for instance {InstanceId} deserialized to null; " +
"returning hash-based comparison without a structured diff",
instanceId);
}
}
catch (JsonException ex)
{
_logger.LogWarning(ex,
"Could not deserialize deployed snapshot for instance {InstanceId}; " +
"returning hash-based comparison without a structured diff",
instanceId);
}
var isStale = deployedRevisionHash != currentHash;
var result = new DeploymentComparisonResult(
instanceId,
deployedRevisionHash,
currentHash,
isStale,
snapshot.DeployedAt,
diff);
return Result<DeploymentComparisonResult>.Success(result);
}
/// <summary>
/// I-1 (latent): returns a copy of <paramref name="config"/> whose
/// <see cref="DataType.List"/> attribute values have been round-tripped through
/// <see cref="AttributeValueCodec.Decode"/> → <see cref="AttributeValueCodec.Encode"/>
/// (native JSON-array form). This normalizes a value deployed in the OLD quoted
/// form (e.g. <c>["10","20"]</c>) to the native form (<c>[10,20]</c>) the current
/// flattener now produces, so the staleness hash and the structured diff do not
/// report a spurious change. Scalar / string attributes are returned unchanged
/// (only <see cref="DataType.List"/> is normalized). A value that cannot be
/// decoded (malformed JSON, bad element, or an unparseable element type) is left
/// as-is — a normalization failure must never break the read-only comparison.
/// </summary>
private ResolvedAttribute NormalizeListAttribute(ResolvedAttribute attr)
{
if (!string.Equals(attr.DataType, nameof(DataType.List), StringComparison.OrdinalIgnoreCase)
|| string.IsNullOrEmpty(attr.Value))
{
return attr;
}
if (!Enum.TryParse<DataType>(attr.ElementDataType, ignoreCase: true, out var elementType)
|| !AttributeValueCodec.IsValidElementType(elementType))
{
return attr;
}
try
{
var normalized = AttributeValueCodec.Encode(
AttributeValueCodec.Decode(attr.Value, DataType.List, elementType));
return normalized == attr.Value ? attr : attr with { Value = normalized };
}
catch (FormatException ex)
{
// Best-effort: a snapshot value that no longer round-trips is left
// untouched rather than aborting the comparison. Logged so an operator
// can investigate the stored value.
_logger.LogWarning(ex,
"Could not normalize List attribute '{Attribute}' in deployed snapshot; " +
"comparing its stored value verbatim",
attr.CanonicalName);
return attr;
}
}
/// <summary>
/// I-1 (latent): applies <see cref="NormalizeListAttribute"/> to every attribute
/// in <paramref name="config"/>, returning the original instance unchanged when
/// no List value needed normalizing (the common scalar-only case).
/// </summary>
private FlattenedConfiguration NormalizeListAttributeValues(FlattenedConfiguration config)
{
if (config.Attributes.Count == 0)
return config;
var normalized = config.Attributes.Select(NormalizeListAttribute).ToList();
var changed = normalized.Where((a, i) => !ReferenceEquals(a, config.Attributes[i])).Any();
return changed ? config with { Attributes = normalized } : config;
}
/// <summary>
/// Returns the current persisted <see cref="DeploymentRecord"/> for
/// the given deployment ID from the configuration database. This is a pure
/// local DB read — it does not contact the site. The query-the-site-before-
/// redeploy reconciliation (design: "Deployment Identity &amp; Idempotency")
/// lives in <see cref="TryReconcileWithSiteAsync"/>, which
/// <see cref="DeployInstanceAsync"/> invokes on the deploy path.
/// </summary>
/// <param name="deploymentId">The unique deployment identifier to look up.</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>A task that resolves to the matching deployment record, or null if none exists.</returns>
public async Task<DeploymentRecord?> GetDeploymentStatusAsync(
string deploymentId,
CancellationToken cancellationToken = default)
{
return await _repository.GetDeploymentByDeploymentIdAsync(deploymentId, cancellationToken);
}
/// <summary>
/// Query-the-site-before-redeploy reconciliation.
///
/// The site query is issued ONLY when a prior <see cref="DeploymentRecord"/>
/// for this instance is stuck <see cref="DeploymentStatus.InProgress"/>, or
/// is <see cref="DeploymentStatus.Failed"/> due to a timeout — the only
/// cases where the site may have applied the config without central
/// learning of it. Fresh first-time deploys and redeploys after a clean
/// prior <see cref="DeploymentStatus.Success"/> skip the extra round-trip.
///
/// Reconciliation: if the site already has the TARGET revision hash, the
/// prior record is marked <see cref="DeploymentStatus.Success"/> (with its
/// <see cref="DeploymentRecord.RevisionHash"/> corrected to the target)
/// and returned (the caller must NOT re-send the
/// deploy). The same post-success side effects as the normal deploy path
/// are applied — instance <see cref="InstanceState.Enabled"/> and a stored
/// <see cref="DeployedConfigSnapshot"/> — so central
/// and site state do not diverge. Otherwise <c>null</c> is returned and the
/// normal deploy proceeds.
///
/// Query failure: if the site is unreachable or the query times out, this
/// returns <c>null</c> (fall through to a normal deploy) — site-side
/// stale-rejection of an older revision hash is the safety net. The deploy
/// is never aborted on a failed query.
/// </summary>
private async Task<DeploymentRecord?> TryReconcileWithSiteAsync(
Instance instance,
string targetRevisionHash,
string configJson,
string currentUser,
CancellationToken cancellationToken)
{
var prior = await _repository.GetCurrentDeploymentStatusAsync(instance.Id, cancellationToken);
if (prior == null || !ShouldQuerySiteBeforeRedeploy(prior))
return null;
DeploymentStateQueryResponse response;
try
{
var siteId = await ResolveSiteIdentifierAsync(instance.SiteId, cancellationToken);
var query = new DeploymentStateQueryRequest(
Guid.NewGuid().ToString("N"), instance.UniqueName, DateTimeOffset.UtcNow);
_logger.LogInformation(
"Querying site {SiteId} for applied deployment state of instance {Instance} " +
"before re-deploy (prior record {DeploymentId} is {Status})",
siteId, instance.UniqueName, prior.DeploymentId, prior.Status);
response = await _communicationService.QueryDeploymentStateAsync(
siteId, query, cancellationToken);
}
catch (Exception ex)
{
// Query failure (site unreachable / timeout): do NOT abort. Fall
// through to a normal deploy; site-side stale-rejection of an older
// revision hash is the safety net.
_logger.LogWarning(ex,
"Site query before re-deploy of instance {Instance} failed; " +
"proceeding with normal deploy (site-side stale-rejection is the safety net)",
instance.UniqueName);
return null;
}
if (response.IsDeployed &&
string.Equals(response.AppliedRevisionHash, targetRevisionHash, StringComparison.Ordinal))
{
// The site already has the target revision — the prior deployment
// actually succeeded. Reconcile the stale record instead of
// re-sending the deploy.
_logger.LogInformation(
"Site already has target revision {RevisionHash} for instance {Instance}; " +
"marking prior deployment record {DeploymentId} Success without re-deploying",
targetRevisionHash, instance.UniqueName, prior.DeploymentId);
prior.Status = DeploymentStatus.Success;
prior.ErrorMessage = null;
prior.CompletedAt = DateTimeOffset.UtcNow;
// The prior record can legitimately carry a
// different (stale) revision hash than the current target. The site
// confirmed it is running the target revision, so the persisted
// record, the audit entry below, and the site must all agree.
prior.RevisionHash = targetRevisionHash;
await _repository.UpdateDeploymentRecordAsync(prior, cancellationToken);
await _repository.SaveChangesAsync(cancellationToken);
NotifyStatusChange(prior);
// A reconciled deployment must perform the
// SAME post-success side effects as the normal deploy path — set
// the instance State to Enabled and store/refresh the deployed
// config snapshot — otherwise the central state machine and the
// deployed-snapshot invariant diverge from what the site is running.
//
// The reconciliation path runs only when the
// prior record is InProgress or timeout-Failed — exactly the cases
// that survive a central failover. The in-memory operation lock is
// lost on failover, so an operator may have legitimately invoked
// Disable on the instance between the original timed-out deploy and
// this redeploy. Disable does not change the deployed config, so the
// site still reports the target revision hash. Reconciliation must
// therefore PRESERVE an intentional Disabled state instead of
// silently flipping it back to Enabled — pass forceEnabledState:
// false so the helper only promotes NotDeployed → Enabled (the
// first-deploy-timed-out case) and leaves an explicit Disabled
// alone.
await ApplyPostSuccessSideEffectsAsync(
instance, prior.DeploymentId, targetRevisionHash, configJson,
forceEnabledState: false, cancellationToken);
// Attribute the audit row to the user driving
// THIS redeploy (the caller of DeployInstanceAsync), not the user
// who issued the original timed-out / stuck deployment. The original
// deployer is preserved in the detail object so forensics can still
// see who launched the run that reconciliation rescued.
await _auditService.LogAsync(currentUser, "DeployReconciled", "Instance",
instance.Id.ToString(), instance.UniqueName,
new
{
DeploymentId = prior.DeploymentId,
RevisionHash = targetRevisionHash,
OriginalDeployer = prior.DeployedBy
},
cancellationToken);
return prior;
}
// Site does not have the target revision (or is not deployed) — proceed
// with the normal deploy.
return null;
}
/// <summary>
/// Runs <see cref="TryReconcileWithSiteAsync"/> for one instance of a BULK
/// deploy, under the batch's short per-reconcile deadline and its
/// once-per-batch reachability circuit-breaker.
///
/// <para>
/// Phase 1 is strictly serial, so an unreachable site makes every instance pay
/// the full communication-layer <c>QueryTimeout</c> (30 s) in turn before a
/// single deploy is sent — 500 instances is roughly four hours, so the CLI's
/// 5-minute bulk-deploy deadline cancels the operation and the next attempt
/// re-arms the identical cost. A bulk redeploy could therefore never converge
/// while the site was down. Two bounds fix that: a short
/// <see cref="DeploymentManagerOptions.SiteDeploymentReconcileTimeout"/> per
/// reconcile, and — after the FIRST timeout — skipping the remaining reconciles
/// for the rest of the batch.
/// </para>
///
/// <para>
/// Skipping loses nothing: a reconcile that times out already returns
/// <see langword="null"/> and the instance proceeds on DB state with the site's
/// own stale-revision rejection as the safety net — which is exactly what every
/// subsequent reconcile against a down site would have produced, one 30-second
/// wait at a time.
/// </para>
///
/// <para>
/// The gate needs no synchronisation: it is touched only from the serial
/// prepare phase.
/// </para>
/// </summary>
/// <param name="instance">Instance being prepared.</param>
/// <param name="targetRevisionHash">Revision hash this deploy would apply.</param>
/// <param name="configJson">Serialized flattened config for the reconciled snapshot.</param>
/// <param name="currentUser">User attributed on the reconciliation audit row.</param>
/// <param name="gate">The batch's reconcile budget and reachability state.</param>
/// <param name="cancellationToken">Cancellation token for the batch.</param>
/// <returns>The reconciled prior record, or <see langword="null"/> to proceed with a normal deploy.</returns>
private async Task<DeploymentRecord?> ReconcileUnderBatchGateAsync(
Instance instance,
string targetRevisionHash,
string configJson,
string currentUser,
BulkReconcileGate gate,
CancellationToken cancellationToken)
{
if (gate.SiteUnreachable)
{
gate.CountSkipped();
return null;
}
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cts.CancelAfter(gate.ReconcileTimeout);
var reconciled = await TryReconcileWithSiteAsync(
instance, targetRevisionHash, configJson, currentUser, cts.Token);
// TryReconcileWithSiteAsync swallows its own query faults (it must — a
// failed query never aborts a deploy), so the deadline is detected on the
// token rather than on an exception. An outer cancellation is the caller
// aborting the whole batch, not evidence about the site.
if (reconciled == null && cts.IsCancellationRequested && !cancellationToken.IsCancellationRequested)
gate.MarkUnreachable();
return reconciled;
}
/// <summary>
/// The site is queried before a re-deploy only when a
/// prior record is stuck <see cref="DeploymentStatus.InProgress"/>, or is
/// <see cref="DeploymentStatus.Failed"/> because the site command timed out
/// (detected via the <see cref="TimeoutFailurePrefix"/> error-message
/// marker). All other prior states skip the query.
/// </summary>
private static bool ShouldQuerySiteBeforeRedeploy(DeploymentRecord prior) =>
prior.Status == DeploymentStatus.InProgress
|| (prior.Status == DeploymentStatus.Failed
&& prior.ErrorMessage != null
&& prior.ErrorMessage.StartsWith(TimeoutFailurePrefix, StringComparison.Ordinal));
/// <summary>
/// Post-success side effects shared by the normal deploy path and the
/// reconciliation path: set the instance
/// <see cref="InstanceState.Enabled"/> and store/refresh the
/// deployed config snapshot. Factored into one helper so the two
/// paths cannot drift.
///
/// <paramref name="forceEnabledState"/> distinguishes
/// the two callers. The normal deploy path passes <c>true</c> — a fresh
/// successful apply legitimately puts the instance into <see cref="InstanceState.Enabled"/>
/// (the documented "Deploy on a Disabled instance also enables it" semantics
/// of <see cref="StateTransitionValidator"/>). The reconciliation path
/// passes <c>false</c>: it is reconciling a *prior* deployment that may
/// have completed before the current operator session (central failover
/// loses the in-memory operation lock, so an operator may have legitimately
/// Disabled the instance in between). On that path we only promote
/// <see cref="InstanceState.NotDeployed"/> → <see cref="InstanceState.Enabled"/>
/// (the first-deploy-timed-out case) and leave an explicit Disabled alone,
/// so reconciliation never silently undoes a Disable.
///
/// Best-effort: the deployment record's terminal <see cref="DeploymentStatus.Success"/>
/// status is already committed by the caller before this runs. A failure
/// here is logged loudly for operator reconciliation but is NOT propagated —
/// it must not flip the already-committed Success record back to Failed.
/// </summary>
private async Task ApplyPostSuccessSideEffectsAsync(
Instance instance,
string deploymentId,
string revisionHash,
string configJson,
bool forceEnabledState,
CancellationToken cancellationToken)
{
try
{
// Update instance state to Enabled on successful deployment.
// On the reconciliation path
// (forceEnabledState=false) only promote NotDeployed → Enabled,
// preserving an intentional Disabled state set between the original
// timed-out deploy and the redeploy.
if (forceEnabledState || instance.State == InstanceState.NotDeployed)
{
instance.State = InstanceState.Enabled;
}
await _repository.UpdateInstanceAsync(instance, cancellationToken);
// Store deployed config snapshot
await StoreDeployedSnapshotAsync(
instance.Id, deploymentId, revisionHash, configJson, cancellationToken);
await _repository.SaveChangesAsync(cancellationToken);
}
catch (Exception postEx)
{
_logger.LogError(postEx,
"Deployment {DeploymentId} for instance {Instance} was applied by the site and " +
"recorded Success, but post-success persistence (instance state / config snapshot) " +
"failed -- central and site state may diverge until reconciled",
deploymentId, instance.UniqueName);
}
}
/// <summary>
/// Write a "&lt;Action&gt;TimedOut" audit entry on
/// behalf of a lifecycle command (Disable / Enable / Delete) whose site
/// round-trip exceeded <see cref="DeploymentManagerOptions.LifecycleCommandTimeout"/>.
///
/// <para>
/// Mirrors the <c>DeployFailed</c> pattern in
/// <see cref="DeployInstanceAsync"/>: the audit write uses
/// <see cref="CancellationToken.None"/> so the operator's outer cancellation
/// (the usual reason this path runs) cannot also prevent the audit row from
/// being persisted. The detail object carries the lifecycle command id, the
/// timeout that fired, and the original exception message so an operator can
/// correlate the audit entry with the UI-surfaced timeout error.
/// </para>
///
/// <para>
/// Wrapped in try/catch — a failed audit write must NOT mask the underlying
/// timeout from the caller; it is logged at Warning so the operator can
/// reconcile but never thrown.
/// </para>
/// </summary>
/// <param name="user">The username who initiated the lifecycle command.</param>
/// <param name="action">The audit action name (<c>DisableTimedOut</c>, <c>EnableTimedOut</c>, or <c>DeleteTimedOut</c>).</param>
/// <param name="instanceId">The numeric instance id, recorded on the audit row.</param>
/// <param name="instanceUniqueName">The instance unique name used as the audit target name.</param>
/// <param name="commandId">The lifecycle command's correlation id, so the audit entry can be matched to logs.</param>
/// <param name="timeoutException">The captured <see cref="TimeoutException"/> or <see cref="OperationCanceledException"/>.</param>
/// <summary>
/// Best-effort audit write for a post-terminal-status action. By the time this
/// runs, the deployment/lifecycle action has already committed its authoritative
/// result; a failed audit write must be logged for operator reconciliation but
/// must NEVER surface to the caller or flip the committed status — "audit-write
/// failure NEVER aborts the user-facing action" (Centralized Audit Log design).
/// Any exception is swallowed and logged at Warning.
/// </summary>
private async Task TryLogAuditAsync(
string user,
string action,
string entityType,
string entityId,
string entityName,
object? afterState,
CancellationToken cancellationToken)
{
try
{
await _auditService.LogAsync(
user, action, entityType, entityId, entityName, afterState, cancellationToken);
}
catch (Exception auditEx)
{
_logger.LogWarning(auditEx,
"Failed to write {Action} audit entry for {EntityType} {EntityName} -- the action " +
"itself succeeded and its committed status is authoritative; audit is best-effort",
action, entityType, entityName);
}
}
private async Task TryLogLifecycleTimeoutAsync(
string user,
string action,
int instanceId,
string instanceUniqueName,
string commandId,
Exception timeoutException)
{
try
{
await _auditService.LogAsync(
user,
action,
"Instance",
instanceId.ToString(),
instanceUniqueName,
new
{
CommandId = commandId,
Deadline = _options.LifecycleCommandTimeout,
Error = timeoutException.Message,
},
CancellationToken.None);
}
catch (Exception auditEx)
{
// A failed audit write must not bury the timeout for the caller —
// just log so an operator can investigate the audit-pipeline issue.
_logger.LogWarning(auditEx,
"Failed to write {Action} audit entry for instance {Instance} (commandId={CommandId})",
action, instanceUniqueName, commandId);
}
}
private async Task StoreDeployedSnapshotAsync(
int instanceId,
string deploymentId,
string revisionHash,
string configJson,
CancellationToken cancellationToken)
{
var existing = await _repository.GetDeployedSnapshotByInstanceIdAsync(instanceId, cancellationToken);
if (existing != null)
{
existing.DeploymentId = deploymentId;
existing.RevisionHash = revisionHash;
existing.ConfigurationJson = configJson;
existing.DeployedAt = DateTimeOffset.UtcNow;
await _repository.UpdateDeployedSnapshotAsync(existing, cancellationToken);
}
else
{
var snapshot = new DeployedConfigSnapshot(deploymentId, revisionHash, configJson)
{
InstanceId = instanceId,
DeployedAt = DateTimeOffset.UtcNow
};
await _repository.AddDeployedSnapshotAsync(snapshot, cancellationToken);
}
}
}
/// <summary>
/// Result of comparing deployed vs template-derived configuration.
/// </summary>
/// <param name="Diff">
/// Structured added/removed/changed detail for
/// attributes, alarms, and scripts. Null only when the deployed snapshot could
/// not be deserialized (corrupt / older schema), in which case
/// <see cref="IsStale"/> still reflects the hash comparison.
/// </param>
public record DeploymentComparisonResult(
int InstanceId,
string DeployedRevisionHash,
string CurrentRevisionHash,
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>
/// One instance a bulk site deployment did not attempt at all — no flatten, no
/// site round-trip, and no state change of any kind.
/// </summary>
/// <param name="InstanceId">Database id of the instance.</param>
/// <param name="UniqueName">The instance's unique name.</param>
/// <param name="Reason">Why it was skipped, phrased for the operator.</param>
public record SkippedInstanceResult(
int InstanceId,
string UniqueName,
string Reason);
/// <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>
/// <param name="SkippedInstances">
/// Instances the batch deliberately did not attempt: the
/// <see cref="InstanceState.Disabled"/> ones, which a bulk push must not silently
/// re-enable, and the up-to-date ones, which a bulk push must not needlessly reset
/// (a redeploy clears static overrides and native alarm state at the site). Each
/// row carries the reason. They are neither successes nor failures, so they are
/// counted in neither total. Additive and nullable so an older client deserialising
/// this summary is unaffected.
/// </param>
public record SiteDeploymentSummary(
string SiteIdentifier,
IReadOnlyList<InstanceDeploymentResult> InstanceResults,
int SuccessCount,
int FailureCount,
IReadOnlyList<SkippedInstanceResult>? SkippedInstances = null);