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.
This commit is contained in:
Joseph Doherty
2026-09-09 22:05:30 -04:00
parent 146a13238f
commit 9a87d7a328
136 changed files with 5441 additions and 648 deletions
@@ -22,7 +22,8 @@ public sealed class PerTargetRedactionOverride
/// whose name matches are replaced with <c>&lt;redacted&gt;</c>. Null (the
/// default) means parameter values are captured verbatim. Only applied to
/// <see cref="ZB.MOM.WW.ScadaBridge.Commons.Types.Enums.AuditChannel.DbOutbound"/>
/// rows.
/// rows. When configured, free-text SQL errors are suppressed because they can echo
/// sensitive parameter values without their names.
/// </summary>
public string? RedactSqlParamsMatching { get; set; }
@@ -31,8 +32,9 @@ public sealed class PerTargetRedactionOverride
/// request/response headers and metadata (status, duration, actor, etc.)
/// but the request and response body strings are omitted
/// (<c>RequestSummary</c> / <c>ResponseSummary</c> are left null). The
/// audit row itself is always emitted — only the body content is suppressed.
/// Null (the default, equivalent to <c>false</c>) means body capture
/// audit row itself is still emitted; exception message/detail are also suppressed
/// because failures may quote body values.
/// The default <c>false</c> means body capture
/// proceeds normally up to <see cref="AuditLogOptions.InboundMaxBytes"/>.
/// </summary>
public bool SkipBodyCapture { get; set; }
@@ -192,6 +192,7 @@ public sealed class ScadaBridgeAuditRedactor : IAuditRedactor, IDisposable
var request = RedactHeaders(d.RequestSummary, opts.HeaderRedactList);
var response = RedactHeaders(d.ResponseSummary, opts.HeaderRedactList);
var errorDetail = d.ErrorDetail;
var errorMessage = d.ErrorMessage;
var extra = d.Extra;
// --- Body-regex stage (also runs BEFORE truncation) -----------
@@ -211,6 +212,7 @@ public sealed class ScadaBridgeAuditRedactor : IAuditRedactor, IDisposable
request = RedactBody(request, bodyRegexes);
response = RedactBody(response, bodyRegexes);
errorDetail = RedactBody(errorDetail, bodyRegexes);
errorMessage = RedactBody(errorMessage, bodyRegexes);
extra = RedactBody(extra, bodyRegexes);
}
@@ -223,6 +225,10 @@ public sealed class ScadaBridgeAuditRedactor : IAuditRedactor, IDisposable
opts, rawEvent.Target, out var sqlParamRegex, out var sqlRedactorUnavailable))
{
request = RedactSqlParameters(request, sqlParamRegex!);
// Parameter policy matches names, whereas SQL exceptions can echo VALUES.
// Free text cannot be safely filtered with a name regex. Keep status/category.
if (errorMessage is not null) errorMessage = "Database operation failed (sensitive SQL error suppressed)";
if (errorDetail is not null) errorDetail = "<redacted: SQL parameter privacy policy>";
}
else if (sqlRedactorUnavailable)
{
@@ -235,6 +241,7 @@ public sealed class ScadaBridgeAuditRedactor : IAuditRedactor, IDisposable
request = TruncateField(request, cap, ref truncated);
response = TruncateField(response, cap, ref truncated);
errorDetail = TruncateField(errorDetail, cap, ref truncated);
errorMessage = TruncateField(errorMessage, cap, ref truncated);
extra = TruncateField(extra, cap, ref truncated);
var rewritten = d with
@@ -242,6 +249,7 @@ public sealed class ScadaBridgeAuditRedactor : IAuditRedactor, IDisposable
RequestSummary = request,
ResponseSummary = response,
ErrorDetail = errorDetail,
ErrorMessage = errorMessage,
Extra = extra,
PayloadTruncated = d.PayloadTruncated || truncated,
};
@@ -1,5 +1,6 @@
using System.CommandLine;
using System.CommandLine.Parsing;
using System.Text.Json;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
namespace ZB.MOM.WW.ScadaBridge.CLI.Commands;
@@ -77,16 +78,31 @@ public static class DeployCommands
// A bulk deploy is N instance round-trips; the default 30 s client
// timeout would abandon the request while the server is still
// applying. Matches the server-side long-running Ask window.
timeout: BulkDeployTimeout);
timeout: BulkDeployTimeout,
onSuccess: json => HandleSiteDeploymentResponse(json, CommandHelpers.ResolveFormat(result, formatOption, CliConfig.Load())));
});
return cmd;
}
/// <summary>
/// Client-side timeout for the bulk site deploy, matching the management
/// service's long-running command Ask window.
/// The server owns the bounded overall bulk deadline; do not abandon it with a shorter client timer.
/// </summary>
internal static readonly TimeSpan BulkDeployTimeout = TimeSpan.FromMinutes(5);
internal static readonly TimeSpan BulkDeployTimeout = Timeout.InfiniteTimeSpan;
internal static int HandleSiteDeploymentResponse(string json, string format)
{
CommandHelpers.HandleResponse(new ManagementResponse(200, json, null, null), format);
try
{
using var payload = JsonDocument.Parse(json);
if (payload.RootElement.TryGetProperty("failureCount", out var failures)
&& failures.TryGetInt32(out var count) && count >= 0)
return count == 0 ? 0 : 1;
}
catch (JsonException) { }
OutputFormatter.WriteError("The server returned an invalid site deployment summary.", "INVALID_RESPONSE");
return 1;
}
private static Command BuildArtifacts(Option<string> urlOption, Option<string> formatOption, Option<string> usernameOption, Option<string> passwordOption)
{
@@ -0,0 +1,6 @@
namespace ZB.MOM.WW.ScadaBridge.CentralUI.SandboxWorker;
internal static class WorkerProgram
{
public static Task Main() => ScriptAnalysis.SandboxWorkerProtocol.RunWorkerAsync();
}
@@ -0,0 +1,5 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup><OutputType>Exe</OutputType><TargetFramework>net10.0</TargetFramework><ImplicitUsings>enable</ImplicitUsings><Nullable>enable</Nullable><TreatWarningsAsErrors>true</TreatWarningsAsErrors></PropertyGroup>
<ItemGroup><FrameworkReference Include="Microsoft.AspNetCore.App" /></ItemGroup>
<ItemGroup><ProjectReference Include="../ZB.MOM.WW.ScadaBridge.CentralUI/ZB.MOM.WW.ScadaBridge.CentralUI.csproj" /></ItemGroup>
</Project>
@@ -0,0 +1,45 @@
using System.Text;
namespace ZB.MOM.WW.ScadaBridge.CentralUI.ScriptAnalysis;
/// <summary>Thread-safe console sink that never retains more than its character budget.</summary>
public sealed class BoundedSandboxWriter(int capacity) : StringWriter
{
private readonly StringBuilder _retained = new(Math.Max(0, capacity));
private readonly object _gate = new();
private long _dropped;
public int RetainedCharacters { get { lock (_gate) return _retained.Length; } }
public long DroppedCharacters { get { lock (_gate) return _dropped; } }
public override void Write(char value) { Span<char> one = stackalloc char[1]; one[0] = value; Write(one); }
public override void Write(string? value) { if (value is not null) Write(value.AsSpan()); }
public override void Write(char[] buffer, int index, int count) => Write(buffer.AsSpan(index, count));
public override void Write(ReadOnlySpan<char> buffer)
{
lock (_gate)
{
var take = Math.Min(buffer.Length, Math.Max(0, capacity - _retained.Length));
_retained.Append(buffer[..take]);
_dropped = Math.Min(long.MaxValue - (buffer.Length - take), _dropped) + buffer.Length - take;
}
}
public override void WriteLine(string? value) { lock (_gate) { Write(value); Write(NewLine); } }
public override void WriteLine() => Write(NewLine);
public override string ToString() { lock (_gate) return _retained.ToString() + (_dropped > 0 ? "\n… (truncated)" : ""); }
}
/// <summary>Rejects serialized data once its byte budget is reached, without retaining the excess.</summary>
public sealed class BoundedSandboxStream(int capacity) : MemoryStream
{
public override void Write(byte[] buffer, int offset, int count) => Write(buffer.AsSpan(offset, count));
public override void Write(ReadOnlySpan<byte> buffer)
{
if (Length + buffer.Length > capacity) throw new InvalidDataException("Sandbox output exceeds its byte budget.");
var bounded = buffer.ToArray();
base.Write(bounded, 0, bounded.Length);
}
public override void WriteByte(byte value) { Span<byte> one = stackalloc byte[1]; one[0] = value; Write(one); }
public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
{ cancellationToken.ThrowIfCancellationRequested(); Write(buffer.Span); return ValueTask.CompletedTask; }
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{ cancellationToken.ThrowIfCancellationRequested(); Write(buffer, offset, count); return Task.CompletedTask; }
}
@@ -0,0 +1,224 @@
using System.Buffers.Binary;
using System.Data.Common;
using System.Diagnostics;
using System.Text.Json;
using Microsoft.Extensions.Caching.Memory;
using Microsoft.Extensions.DependencyInjection;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Communication;
namespace ZB.MOM.WW.ScadaBridge.CentralUI.ScriptAnalysis;
/// <summary>Bounded private IPC for disposable, killable Test Run processes.</summary>
public static class SandboxWorkerProtocol
{
private const int FrameLimit = 1024 * 1024;
private static readonly SemaphoreSlim Admission = new(4, 4);
private static readonly JsonSerializerOptions Json = new() { IgnoreReadOnlyProperties = true };
private sealed record Frame(string Kind, string? Method, JsonElement Data, string? Error = null);
private sealed record Start(SandboxRunRequest Request, bool External, bool Database, bool Bound);
private sealed record ConnectionInfo(string Provider, string ConnectionString, bool Open);
/// <summary>Runs one child under a deadline; a slot remains charged until confirmed process exit.</summary>
public static async Task<SandboxRunResult> RunAsync(SandboxRunRequest request, IServiceProvider services,
ISharedScriptCatalog catalog, CancellationToken caller)
=> await RunCoreAsync(request, services, catalog, caller);
internal static async Task<SandboxRunResult> RunCoreAsync(SandboxRunRequest request, IServiceProvider services,
ISharedScriptCatalog catalog, CancellationToken caller, Action<int>? started = null)
{
var seconds = Math.Clamp(request.TimeoutSeconds ?? 5, 1, 10);
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(caller);
deadline.CancelAfter(TimeSpan.FromSeconds(seconds));
var clock = Stopwatch.StartNew();
var acquired = false;
Process? process = null;
Task? stderr = null;
CancellationTokenRegistration terminateOnDeadline = default;
try
{
await Admission.WaitAsync(deadline.Token); acquired = true;
ISandboxInstanceGateway? gateway = null;
if (request.Kind != ScriptKind.InboundApi && !string.IsNullOrWhiteSpace(request.BindInstanceUniqueName))
{
var locator = services.GetService<IInstanceLocator>();
var comms = services.GetService<CommunicationService>();
var site = locator is null ? null : await locator.GetSiteIdForInstanceAsync(request.BindInstanceUniqueName, deadline.Token);
if (site is null || comms is null) throw new ScriptSandboxException("Cannot bind instance: cross-site communication or deployment is unavailable.");
gateway = new SandboxInstanceGateway(comms, site, request.BindInstanceUniqueName, deadline.Token);
}
var worker = Path.Combine(AppContext.BaseDirectory, "ZB.MOM.WW.ScadaBridge.CentralUI.SandboxWorker.dll");
if (!File.Exists(worker)) throw new InvalidOperationException("Sandbox worker is not installed.");
var start = new ProcessStartInfo("dotnet") { UseShellExecute = false, RedirectStandardInput = true,
RedirectStandardOutput = true, RedirectStandardError = true, CreateNoWindow = true };
start.ArgumentList.Add(worker);
// A finite heap bounds accidental allocation in return serialization and script code.
start.Environment["DOTNET_GCHeapHardLimit"] = "20000000"; // 512 MiB, hexadecimal runtime setting.
process = Process.Start(start) ?? throw new InvalidOperationException("Could not start sandbox worker.");
terminateOnDeadline = deadline.Token.Register(static state =>
{
var child = (Process)state!;
try { if (!child.HasExited) child.Kill(entireProcessTree: true); }
catch (InvalidOperationException) { /* Concurrent normal process exit. */ }
}, process);
started?.Invoke(process.Id);
stderr = DrainAsync(process.StandardError.BaseStream);
await WriteAsync(process.StandardInput.BaseStream, new Frame("start", null,
Encode(new Start(request, services.GetService<IExternalSystemClient>() is not null,
services.GetService<IDatabaseGateway>() is not null, gateway is not null))), deadline.Token);
while (true)
{
var frame = await ReadAsync(process.StandardOutput.BaseStream, deadline.Token);
if (frame.Kind == "result")
return frame.Data.Deserialize<SandboxRunResult>(Json) ?? throw new InvalidDataException("Missing sandbox result.");
if (frame.Kind != "call") throw new InvalidDataException("Unexpected sandbox message.");
deadline.Token.ThrowIfCancellationRequested();
Frame reply;
try
{
var value = await DispatchAsync(frame.Method!, frame.Data, services, catalog, gateway, deadline.Token);
deadline.Token.ThrowIfCancellationRequested();
reply = new Frame("reply", null, Encode(value));
}
catch (Exception ex) when (ex is not OperationCanceledException)
{ reply = new Frame("reply", null, Encode(null), ex is ScriptSandboxException ? ex.Message : "Sandbox gateway request failed."); }
await WriteAsync(process.StandardInput.BaseStream, reply, deadline.Token);
}
}
catch (OperationCanceledException)
{ return new(false, null, null, "", $"Script execution exceeded the {seconds}-second sandbox timeout or was canceled.", SandboxErrorKind.Timeout, clock.ElapsedMilliseconds, null); }
catch (ScriptSandboxException ex)
{ return new(false, null, null, "", ex.Message, SandboxErrorKind.SandboxLimitation, clock.ElapsedMilliseconds, null); }
catch (Exception)
{ return new(false, null, null, "", "Sandbox worker exited or exceeded its resource budget.", SandboxErrorKind.RuntimeError, clock.ElapsedMilliseconds, null); }
finally
{
deadline.Cancel(); // Stop accepting/brokering new operations before termination.
terminateOnDeadline.Dispose();
if (process is not null)
{
try { if (!process.HasExited) process.Kill(entireProcessTree: true); }
catch (InvalidOperationException) when (process.HasExited) { }
await process.WaitForExitAsync(CancellationToken.None);
if (stderr is not null) await stderr;
process.Dispose();
}
if (acquired) Admission.Release();
}
}
private static async Task DrainAsync(Stream stream)
{ var buffer = new byte[4096]; while (await stream.ReadAsync(buffer) != 0) { } }
private static async Task<object?> DispatchAsync(string method, JsonElement args, IServiceProvider services,
ISharedScriptCatalog catalog, ISandboxInstanceGateway? gateway, CancellationToken ct)
{
string Text(int i) => args[i].GetString()!;
IReadOnlyDictionary<string, object?>? Parameters(int i) => args[i].ValueKind == JsonValueKind.Null ? null
: args[i].EnumerateObject().ToDictionary(p => p.Name, p => ScriptAnalysisService.JsonElementToObject(p.Value));
switch (method)
{
case "shared": return await catalog.GetByNameAsync(Text(0), ct);
case "external": return await services.GetRequiredService<IExternalSystemClient>().CallAsync(Text(0), Text(1), Parameters(2), ct);
case "cachedExternal": return await services.GetRequiredService<IExternalSystemClient>().CachedCallAsync(Text(0), Text(1), Parameters(2), Text(3), ct);
case "cachedDb": return await services.GetRequiredService<IDatabaseGateway>().CachedWriteAsync(Text(0), Text(1), Parameters(2), Text(3), ct);
case "connection":
var source = services.GetRequiredService<IDatabaseGateway>() as ISandboxDatabaseConnectionSource
?? throw new ScriptSandboxException("Database worker handoff is unavailable for this provider.");
return new ConnectionInfo("Microsoft.Data.SqlClient.SqlConnection",
await source.GetSandboxConnectionStringAsync(Text(0), ct), true);
case "get": return await (gateway ?? throw new ScriptSandboxException("Instance not bound.")).GetAttributeAsync(Text(0), ct);
case "set": await gateway!.SetAttributeAsync(Text(0), Text(1), ct); return null;
case "script": return await gateway!.CallScriptAsync(Text(0), Parameters(1), ct);
case "wait": return await gateway!.WaitForAttributeAsync(Text(0), args[1].GetString(), TimeSpan.FromTicks(args[2].GetInt64()), args[3].GetBoolean(), ct);
default: throw new ScriptSandboxException("Unknown sandbox operation.");
}
}
private static JsonElement Encode(object? value)
{
using var buffer = new BoundedSandboxStream(FrameLimit - 4096);
JsonSerializer.Serialize(buffer, value, Json);
using var document = JsonDocument.Parse(buffer.GetBuffer().AsMemory(0, (int)buffer.Length));
return document.RootElement.Clone();
}
private static async Task WriteAsync(Stream stream, Frame frame, CancellationToken ct)
{
using var bytes = new BoundedSandboxStream(FrameLimit);
JsonSerializer.Serialize(bytes, frame, Json);
var prefix = new byte[4]; BinaryPrimitives.WriteInt32LittleEndian(prefix, (int)bytes.Length);
await stream.WriteAsync(prefix, ct);
await stream.WriteAsync(bytes.GetBuffer().AsMemory(0, (int)bytes.Length), ct);
await stream.FlushAsync(ct);
}
private static async Task<Frame> ReadAsync(Stream stream, CancellationToken ct)
{
var prefix = new byte[4]; await stream.ReadExactlyAsync(prefix, ct);
var length = BinaryPrimitives.ReadInt32LittleEndian(prefix);
if (length is < 1 or > FrameLimit) throw new InvalidDataException("Sandbox frame budget exceeded.");
var data = new byte[length]; await stream.ReadExactlyAsync(data, ct);
return JsonSerializer.Deserialize<Frame>(data, Json) ?? throw new InvalidDataException("Invalid sandbox frame.");
}
/// <summary>Worker executable entry. Protocol streams are captured before script console routing.</summary>
public static async Task RunWorkerAsync()
{
var input = Console.OpenStandardInput(); var output = Console.OpenStandardOutput();
var first = await ReadAsync(input, CancellationToken.None);
var start = first.Data.Deserialize<Start>(Json) ?? throw new InvalidDataException("Missing worker request.");
var rpc = new Rpc(input, output);
var services = new ServiceCollection();
if (start.External) services.AddSingleton<IExternalSystemClient>(rpc);
if (start.Database) services.AddSingleton<IDatabaseGateway>(rpc);
if (start.Bound) services.AddSingleton<ISandboxInstanceGateway>(rpc);
using var provider = services.BuildServiceProvider();
using var cache = new MemoryCache(new MemoryCacheOptions());
var runner = new ScriptAnalysisService(rpc, cache, provider);
var result = await runner.RunInWorkerAsync(start.Request, CancellationToken.None);
await WriteAsync(output, new Frame("result", null, Encode(result)), CancellationToken.None);
}
private sealed class Rpc(Stream input, Stream output) : ISharedScriptCatalog, IExternalSystemClient, IDatabaseGateway, ISandboxInstanceGateway
{
private readonly SemaphoreSlim _gate = new(1);
private async Task<JsonElement> Call(string method, object?[] args, CancellationToken ct)
{
await _gate.WaitAsync(ct);
try
{
await WriteAsync(output, new Frame("call", method, Encode(args)), ct);
var reply = await ReadAsync(input, ct);
if (reply.Error is not null) throw new ScriptSandboxException(reply.Error);
return reply.Data;
}
finally { _gate.Release(); }
}
public Task<IReadOnlyList<ScriptShape>> GetShapesAsync() => Task.FromResult<IReadOnlyList<ScriptShape>>([]);
public async Task<SharedScriptSource?> GetByNameAsync(string name, CancellationToken cancellationToken = default)
=> (await Call("shared", [name], cancellationToken)).Deserialize<SharedScriptSource>(Json);
public async Task<ExternalCallResult> CallAsync(string systemName, string methodName, IReadOnlyDictionary<string, object?>? parameters = null, CancellationToken cancellationToken = default)
=> (await Call("external", [systemName, methodName, parameters], cancellationToken)).Deserialize<ExternalCallResult>(Json)!;
public async Task<ExternalCallResult> CachedCallAsync(string systemName, string methodName, IReadOnlyDictionary<string, object?>? parameters = null, string? originInstanceName = null, CancellationToken cancellationToken = default, TrackedOperationId? trackedOperationId = null, Guid? executionId = null, string? sourceScript = null, Guid? parentExecutionId = null)
=> (await Call("cachedExternal", [systemName, methodName, parameters, originInstanceName], cancellationToken)).Deserialize<ExternalCallResult>(Json)!;
public async Task<ExternalCallResult> CachedWriteAsync(string connectionName, string sql, IReadOnlyDictionary<string, object?>? parameters = null, string? originInstanceName = null, CancellationToken cancellationToken = default, TrackedOperationId? trackedOperationId = null, Guid? executionId = null, string? sourceScript = null, Guid? parentExecutionId = null)
=> (await Call("cachedDb", [connectionName, sql, parameters, originInstanceName], cancellationToken)).Deserialize<ExternalCallResult>(Json)!;
public async Task<DbConnection> GetConnectionAsync(string connectionName, CancellationToken cancellationToken = default)
{
var info = (await Call("connection", [connectionName], cancellationToken)).Deserialize<ConnectionInfo>(Json)!;
DbConnection connection = info.Provider switch {
"Microsoft.Data.SqlClient.SqlConnection" => new Microsoft.Data.SqlClient.SqlConnection(info.ConnectionString),
_ => throw new ScriptSandboxException("Unsupported database provider.") };
if (info.Open) await connection.OpenAsync(cancellationToken);
return connection;
}
public async Task<object?> GetAttributeAsync(string canonicalName, CancellationToken ct)
=> ScriptAnalysisService.JsonElementToObject(await Call("get", [canonicalName], ct));
public async Task SetAttributeAsync(string canonicalName, string value, CancellationToken ct) => await Call("set", [canonicalName, value], ct);
public async Task<object?> CallScriptAsync(string canonicalScriptName, IReadOnlyDictionary<string, object?>? parameters, CancellationToken ct)
=> ScriptAnalysisService.JsonElementToObject(await Call("script", [canonicalScriptName, parameters], ct));
public async Task<WaitResult> WaitForAttributeAsync(string canonicalName, string? targetValueEncoded, TimeSpan timeout, bool requireGoodQuality, CancellationToken ct)
=> (await Call("wait", [canonicalName, targetValueEncoded, timeout.Ticks, requireGoodQuality], ct)).Deserialize<WaitResult>(Json)!;
}
}
@@ -189,20 +189,9 @@ public class ScriptAnalysisService
private const int SandboxMaxCallSharedDepth = 16;
/// <summary>
/// Process-wide cap on concurrently executing Test Runs (arch-review WP2.4).
/// <para>
/// A sandbox run occupies a thread-pool thread for its whole duration — the
/// <c>SandboxScriptHost</c> attribute accessors are synchronous by contract and block on
/// cross-site I/O (<c>GetAwaiter().GetResult()</c>), so the thread is <em>parked</em>, not
/// merely busy, for up to the 10s sandbox timeout. Unbounded, a handful of Designers
/// each hitting Run on a slow-bound script could park enough pool threads to starve the
/// Blazor circuits sharing this process. The gate bounds that: excess runs wait for a
/// slot instead of taking a thread, and the wait is charged to the caller's own timeout.
/// </para>
/// <para>
/// Static because <c>ScriptAnalysisService</c> is registered scoped — a per-instance gate
/// would be per-request and bound nothing.
/// </para>
/// Secondary guard inside the disposable worker. Parent admission and process
/// termination are enforced by <see cref="SandboxWorkerProtocol"/>; synchronous
/// script helper calls never occupy a central Blazor thread.
/// </summary>
private static readonly SemaphoreSlim SandboxRunGate =
new(SandboxMaxConcurrentRuns, SandboxMaxConcurrentRuns);
@@ -214,7 +203,7 @@ public class ScriptAnalysisService
private const int SandboxMaxConcurrentRuns = 4;
/// <summary>
/// Compiles and runs a script in the central process. The globals surface
/// Compiles and runs a script in a disposable worker process. The globals surface
/// depends on <see cref="SandboxRunRequest.Kind"/>: template and shared
/// scripts run against <see cref="SandboxScriptHost"/>, inbound API method
/// scripts against <see cref="SandboxInboundScriptHost"/>.
@@ -242,7 +231,11 @@ public class ScriptAnalysisService
/// <param name="request">The sandbox run request including code, kind, parameters, and optional instance binding.</param>
/// <param name="ct">Cancellation token for the caller; combined with an internal timeout CTS.</param>
/// <returns>A <see cref="SandboxRunResult"/> describing the outcome, return value, console output, and any error.</returns>
public virtual async Task<SandboxRunResult> RunInSandboxAsync(SandboxRunRequest request, CancellationToken ct)
public virtual Task<SandboxRunResult> RunInSandboxAsync(SandboxRunRequest request, CancellationToken ct)
=> SandboxWorkerProtocol.RunAsync(request, _services, _sharedScripts, ct);
/// <summary>Execution core invoked only by the disposable worker executable.</summary>
public async Task<SandboxRunResult> RunInWorkerAsync(SandboxRunRequest request, CancellationToken ct)
{
if (string.IsNullOrWhiteSpace(request.Code))
{
@@ -294,7 +287,7 @@ public class ScriptAnalysisService
// Trust-model gate: the documented forbidden-API set is
// enforced HERE, before execution — not merely surfaced as an editor hint.
// Without this, a Design-role user could run arbitrary file/process/
// reflection/network code in the central host process.
// reflection/network code in the worker process.
var trustViolations = EnforceTrustModel(script.GetCompilation());
if (trustViolations.Count > 0)
{
@@ -313,9 +306,9 @@ public class ScriptAnalysisService
// Optional instance binding: when the Test Run targets a deployed
// instance, Instance.GetAttribute/SetAttribute/CallScript and the
// Attributes/Children/Parent accessors route to it cross-site.
ISandboxInstanceGateway? instanceGateway = null;
var instanceLabel = "test-run";
if (request.Kind != ScriptKind.InboundApi
ISandboxInstanceGateway? instanceGateway = _services.GetService<ISandboxInstanceGateway>();
var instanceLabel = request.BindInstanceUniqueName ?? "test-run";
if (instanceGateway is null && request.Kind != ScriptKind.InboundApi
&& !string.IsNullOrWhiteSpace(request.BindInstanceUniqueName))
{
var bindName = request.BindInstanceUniqueName.Trim();
@@ -457,7 +450,7 @@ public class ScriptAnalysisService
// restore Console.Out while the other is still writing. SandboxConsoleCapture
// installs routing writers once and scopes capture to this call-tree only.
var (captureOut, captureError) = SandboxConsoleCapture.Install();
var captured = new StringWriter();
var captured = new BoundedSandboxWriter(SandboxMaxConsoleChars);
using var outScope = captureOut.BeginCapture(captured);
using var errorScope = captureError.BeginCapture(captured);
@@ -485,7 +478,7 @@ public class ScriptAnalysisService
Success: true,
ReturnValueJson: returnJson,
ReturnTypeName: returnType,
ConsoleOutput: TruncateConsole(captured.ToString()),
ConsoleOutput: captured.ToString(),
Error: null,
ErrorKind: SandboxErrorKind.None,
DurationMs: stopwatch.ElapsedMilliseconds,
@@ -495,14 +488,14 @@ public class ScriptAnalysisService
{
stopwatch.Stop();
return new SandboxRunResult(false, null, null,
TruncateConsole(captured.ToString()), sandboxEx.Message,
captured.ToString(), sandboxEx.Message,
SandboxErrorKind.SandboxLimitation, stopwatch.ElapsedMilliseconds, null);
}
catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested)
{
stopwatch.Stop();
return new SandboxRunResult(false, null, null,
TruncateConsole(captured.ToString()),
captured.ToString(),
$"Script execution exceeded the {timeoutSeconds}-second sandbox timeout.",
SandboxErrorKind.Timeout, stopwatch.ElapsedMilliseconds, null);
}
@@ -513,11 +506,11 @@ public class ScriptAnalysisService
if (inner is ScriptSandboxException sx)
{
return new SandboxRunResult(false, null, null,
TruncateConsole(captured.ToString()), sx.Message,
captured.ToString(), sx.Message,
SandboxErrorKind.SandboxLimitation, stopwatch.ElapsedMilliseconds, null);
}
return new SandboxRunResult(false, null, null,
TruncateConsole(captured.ToString()),
captured.ToString(),
$"{inner.GetType().Name}: {inner.Message}",
SandboxErrorKind.RuntimeError, stopwatch.ElapsedMilliseconds, null);
}
@@ -542,7 +535,7 @@ public class ScriptAnalysisService
return result;
}
private static object? JsonElementToObject(JsonElement element)
internal static object? JsonElementToObject(JsonElement element)
{
return element.ValueKind switch
{
@@ -561,25 +554,17 @@ public class ScriptAnalysisService
private static (string Json, string TypeName) SerializeReturn(object? value)
{
if (value == null) return ("null", "null");
var typeName = value.GetType().Name;
var typeName = value?.GetType().FullName ?? "null";
try
{
var json = JsonSerializer.Serialize(value, SandboxReturnJsonOptions);
if (json.Length > SandboxMaxReturnJsonChars)
json = json[..SandboxMaxReturnJsonChars] + "\n… (truncated)";
return (json, typeName);
if (value is string text && text.Length > SandboxMaxReturnJsonChars)
return ("\"<return value truncated>\"", typeName);
using var buffer = new BoundedSandboxStream(SandboxMaxReturnJsonChars);
JsonSerializer.Serialize(buffer, value, SandboxReturnJsonOptions);
return (Encoding.UTF8.GetString(buffer.GetBuffer(), 0, (int)buffer.Length), typeName);
}
catch (Exception ex)
{
return ($"\"<unserializable: {ex.Message}>\"", typeName);
}
}
private static string TruncateConsole(string text)
{
if (text.Length <= SandboxMaxConsoleChars) return text;
return text[..SandboxMaxConsoleChars] + "\n… (truncated)";
catch (InvalidDataException) { return ("\"<return value truncated>\"", typeName); }
catch (Exception) { return ("\"<unserializable>\"", typeName); }
}
private DiagnoseResponse Cache(string key, DiagnoseResponse value)
@@ -55,7 +55,7 @@ public static class ServiceCollectionExtensions
// racing AuditFilterBar's site enumeration on the shared scoped context.
services.AddScoped<IAuditLogQueryService>(sp => new AuditLogQueryService(
sp.GetRequiredService<IServiceScopeFactory>(),
sp.GetRequiredService<ICentralHealthAggregator>()));
sp.GetRequiredService<ICentralHealthAggregator>(), sp.GetRequiredService<AuthenticationStateProvider>()));
// Server-side streaming CSV export.
// Backs the Audit Log page's Export button via GET /api/centralui/audit/export.
@@ -1,3 +1,6 @@
using Microsoft.AspNetCore.Components.Authorization;
using ZB.MOM.WW.ScadaBridge.CentralUI.Auth;
using ZB.MOM.WW.ScadaBridge.Security;
using Microsoft.Extensions.DependencyInjection;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
@@ -45,6 +48,7 @@ public sealed class AuditLogQueryService : IAuditLogQueryService
// Production path: open a fresh scope per operation. Null in the test-seam ctor.
private readonly IServiceScopeFactory? _scopeFactory;
private readonly AuthenticationStateProvider? _auth;
// Test seam: a directly-injected repository whose lifetime the test owns.
// Null in the production ctor.
@@ -68,9 +72,10 @@ public sealed class AuditLogQueryService : IAuditLogQueryService
/// <param name="healthAggregator">Central health aggregator for KPI backlog data.</param>
public AuditLogQueryService(
IServiceScopeFactory scopeFactory,
ICentralHealthAggregator healthAggregator)
ICentralHealthAggregator healthAggregator, AuthenticationStateProvider? auth = null)
{
_scopeFactory = scopeFactory ?? throw new ArgumentNullException(nameof(scopeFactory));
_auth = auth;
_healthAggregator = healthAggregator ?? throw new ArgumentNullException(nameof(healthAggregator));
}
@@ -171,7 +176,12 @@ public sealed class AuditLogQueryService : IAuditLogQueryService
// page's auto-load never shares the circuit-scoped context.
await using var scope = _scopeFactory!.CreateAsyncScope();
var repository = scope.ServiceProvider.GetRequiredService<IAuditLogRepository>();
return await repository.GetExecutionTreeAsync(executionId, ct);
if (_auth is null) throw new InvalidOperationException("Audit tree caller authentication is not configured.");
var principal = (await _auth.GetAuthoritativeStateAsync()).User;
if (principal.Identity?.IsAuthenticated != true || !principal.FindAll(JwtTokenService.RoleClaimType).Any(c => AuthorizationPolicies.OperationalAuditRoles.Contains(c.Value))) throw new UnauthorizedAccessException();
var sites = principal.FindAll(JwtTokenService.SiteIdClaimType).Select(c => c.Value).ToArray();
var permitted = principal.HasClaim(JwtTokenService.RoleClaimType, Roles.Administrator) || sites.Length == 0 ? null : sites;
return await repository.GetExecutionTreeAsync(executionId, ct, permitted);
}
/// <inheritdoc/>
@@ -0,0 +1,5 @@
namespace ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances;
/// <summary>A database-enforced global instance name conflict, including concurrent creation.</summary>
public sealed class InstanceNameConflictException(Exception innerException)
: InvalidOperationException("An instance with this unique name already exists. Choose a different name.", innerException);
@@ -294,7 +294,7 @@ public interface IAuditLogRepository
/// <returns>A task that resolves to the full execution tree rooted at the topmost ancestor, one node per distinct execution.</returns>
Task<IReadOnlyList<ExecutionTreeNode>> GetExecutionTreeAsync(
Guid executionId,
CancellationToken ct = default);
CancellationToken ct = default, IReadOnlyList<string>? permittedSiteIds = null);
/// <summary>
/// Flow Debugger Replay (design 2026-09-05 §7.1): every row of the execution tree that
@@ -309,7 +309,7 @@ public interface IAuditLogRepository
/// <returns>The canonical audit events that make up the flow, oldest first.</returns>
Task<IReadOnlyList<AuditEvent>> GetTraceEventsAsync(
Guid executionId,
CancellationToken ct = default);
CancellationToken ct = default, IReadOnlyList<string>? permittedSiteIds = null);
/// <summary>
/// Returns the distinct, non-null <c>SourceNode</c> values present in the
@@ -0,0 +1,6 @@
namespace ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
/// <summary>Private worker handoff for an existing configured SQL connection. Never log or expose the returned secret.</summary>
public interface ISandboxDatabaseConnectionSource
{
Task<string> GetSandboxConnectionStringAsync(string connectionName, CancellationToken cancellationToken);
}
@@ -0,0 +1,11 @@
using System.Collections;
using ZB.MOM.WW.Audit;
namespace ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
/// <summary>Bounded replay data with independently retained chain-root and truncation metadata.</summary>
public sealed record TraceEventBatch(Guid RootExecutionId, IReadOnlyList<AuditEvent> Events, bool Truncated) : IReadOnlyList<AuditEvent>
{
public AuditEvent this[int index] => Events[index];
public int Count => Events.Count;
public IEnumerator<AuditEvent> GetEnumerator() => Events.GetEnumerator();
IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
}
@@ -58,7 +58,7 @@ public class InstanceConfiguration : IEntityTypeConfiguration<Instance>
.HasForeignKey(o => o.InstanceId)
.OnDelete(DeleteBehavior.Cascade);
builder.HasIndex(i => new { i.SiteId, i.UniqueName }).IsUnique();
builder.HasIndex(i => i.UniqueName).IsUnique();
}
}
@@ -49,11 +49,9 @@ public class DataConnectionConfiguration : IEntityTypeConfiguration<DataConnecti
.IsRequired()
.HasMaxLength(50);
builder.Property(d => d.PrimaryConfiguration)
.HasMaxLength(4000);
builder.Property(d => d.PrimaryConfiguration); // Protected payload may exceed its plaintext size.
builder.Property(d => d.BackupConfiguration)
.HasMaxLength(4000);
builder.Property(d => d.BackupConfiguration);
builder.Property(d => d.FailoverRetryCount)
.IsRequired()
@@ -0,0 +1,18 @@
using Microsoft.AspNetCore.DataProtection;
using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase;
/// <summary>Versioned protection of complete protocol JSON, including nested credentials.</summary>
public sealed class DataConnectionSecretConverter(IDataProtector protector) : ValueConverter<string?, string?>(
value => value == null ? null : Protect(protector, value),
value => value == null ? null : Unprotect(protector, value))
{
internal const string Prefix = "dp:protocol:v1:";
internal const string Purpose = EncryptedStringConverter.ProtectorPurpose + ".DataConnection";
internal static string Protect(IDataProtector protector, string plaintext) => Prefix + protector.Protect(plaintext);
internal static string Unprotect(IDataProtector protector, string stored)
=> stored.StartsWith(Prefix, StringComparison.Ordinal)
? protector.Unprotect(stored[Prefix.Length..])
: throw new InvalidOperationException("Protocol configuration requires the startup encryption upgrade before it can be read.");
}
@@ -0,0 +1,79 @@
using System.Data;
using Microsoft.EntityFrameworkCore;
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase;
/// <summary>Idempotent, bounded startup upgrade of existing plaintext protocol configuration.</summary>
public static class DataConnectionSecretUpgrade
{
/// <summary>Protects both configuration columns before repositories may materialize them.</summary>
public static async Task UpgradeAsync(ScadaBridgeDbContext db, CancellationToken cancellationToken = default)
{
var connection = db.Database.GetDbConnection();
var opened = connection.State != ConnectionState.Open;
if (opened) await connection.OpenAsync(cancellationToken);
try
{
var lastId = 0;
while (true)
{
var rows = new List<(int Id, string? Primary, string? Backup)>();
await using (var read = connection.CreateCommand())
{
var sqlServer = db.Database.IsSqlServer();
read.CommandText = sqlServer
? "SELECT TOP (256) Id, PrimaryConfiguration, BackupConfiguration FROM DataConnections WHERE Id > @last ORDER BY Id"
: "SELECT Id, PrimaryConfiguration, BackupConfiguration FROM DataConnections WHERE Id > @last ORDER BY Id LIMIT 256";
Add(read, "@last", lastId);
await using var reader = await read.ExecuteReaderAsync(cancellationToken);
while (await reader.ReadAsync(cancellationToken))
rows.Add((reader.GetInt32(0), reader.IsDBNull(1) ? null : reader.GetString(1), reader.IsDBNull(2) ? null : reader.GetString(2)));
}
if (rows.Count == 0) break;
foreach (var row in rows)
{
lastId = row.Id;
string? Protect(string? value)
{
if (value is null) return null;
if (value.StartsWith(DataConnectionSecretConverter.Prefix, StringComparison.Ordinal))
{
_ = DataConnectionSecretConverter.Unprotect(db.DataConnectionProtector, value);
return value;
}
return DataConnectionSecretConverter.Protect(db.DataConnectionProtector, value);
}
var primary = Protect(row.Primary);
var backup = Protect(row.Backup);
if (primary == row.Primary && backup == row.Backup) continue;
await using var update = connection.CreateCommand();
// Compare-and-swap avoids overwriting a concurrent upgrade or configuration edit.
update.CommandText = """
UPDATE DataConnections SET PrimaryConfiguration = @primary, BackupConfiguration = @backup
WHERE Id = @id
AND (PrimaryConfiguration = @oldPrimary OR (PrimaryConfiguration IS NULL AND @oldPrimary IS NULL))
AND (BackupConfiguration = @oldBackup OR (BackupConfiguration IS NULL AND @oldBackup IS NULL))
""";
if (db.Database.IsSqlServer())
update.CommandText = update.CommandText
.Replace("PrimaryConfiguration = @oldPrimary", "PrimaryConfiguration COLLATE Latin1_General_100_BIN2 = @oldPrimary COLLATE Latin1_General_100_BIN2", StringComparison.Ordinal)
.Replace("BackupConfiguration = @oldBackup", "BackupConfiguration COLLATE Latin1_General_100_BIN2 = @oldBackup COLLATE Latin1_General_100_BIN2", StringComparison.Ordinal);
Add(update, "@id", row.Id);
Add(update, "@primary", primary); Add(update, "@backup", backup);
Add(update, "@oldPrimary", row.Primary); Add(update, "@oldBackup", row.Backup);
await update.ExecuteNonQueryAsync(cancellationToken);
}
}
}
finally { if (opened) await connection.CloseAsync(); }
}
private static void Add(System.Data.Common.DbCommand command, string name, object? value)
{
var parameter = command.CreateParameter();
parameter.ParameterName = name;
parameter.Value = value ?? DBNull.Value;
if (value is not int) { parameter.DbType = DbType.String; parameter.Size = -1; }
command.Parameters.Add(parameter);
}
}
@@ -48,6 +48,8 @@ public static class MigrationHelper
}
}
await DataConnectionSecretUpgrade.UpgradeAsync(dbContext, cancellationToken);
// Safety-net normalizer: rewrite any already-persisted List attribute values from the
// old array-of-strings JSON form to the new native-typed form. Idempotent and never
// aborts startup for data reasons (per-row skip + log). Safe even if both central
@@ -0,0 +1,98 @@
using Microsoft.EntityFrameworkCore.Migrations;
#nullable disable
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Migrations
{
/// <inheritdoc />
public partial class EnforceGlobalInstanceNamesAndProtectProtocolConfiguration : Migration
{
/// <inheritdoc />
protected override void Up(MigrationBuilder migrationBuilder)
{
migrationBuilder.Sql("""
IF EXISTS (SELECT UniqueName FROM Instances GROUP BY UniqueName HAVING COUNT(*) > 1)
THROW 51002, 'Duplicate global instance names exist. Inventory Instances grouped by UniqueName and resolve ownership before applying this upgrade.', 1;
""");
migrationBuilder.DropIndex(
name: "IX_Instances_SiteId_UniqueName",
table: "Instances");
migrationBuilder.AlterColumn<string>(
name: "PrimaryConfiguration",
table: "DataConnections",
type: "nvarchar(max)",
nullable: true,
oldClrType: typeof(string),
oldType: "nvarchar(4000)",
oldMaxLength: 4000,
oldNullable: true);
migrationBuilder.AlterColumn<string>(
name: "BackupConfiguration",
table: "DataConnections",
type: "nvarchar(max)",
nullable: true,
oldClrType: typeof(string),
oldType: "nvarchar(4000)",
oldMaxLength: 4000,
oldNullable: true);
migrationBuilder.CreateIndex(
name: "IX_Instances_SiteId",
table: "Instances",
column: "SiteId");
migrationBuilder.CreateIndex(
name: "IX_Instances_UniqueName",
table: "Instances",
column: "UniqueName",
unique: true);
}
/// <inheritdoc />
protected override void Down(MigrationBuilder migrationBuilder)
{
migrationBuilder.Sql("""
IF EXISTS (SELECT 1 FROM DataConnections
WHERE PrimaryConfiguration LIKE 'dp:protocol:v1:%' OR BackupConfiguration LIKE 'dp:protocol:v1:%')
THROW 51003, 'Protocol configurations are protected. Restore a coordinated pre-upgrade backup before downgrading application and schema.', 1;
""");
migrationBuilder.DropIndex(
name: "IX_Instances_SiteId",
table: "Instances");
migrationBuilder.DropIndex(
name: "IX_Instances_UniqueName",
table: "Instances");
migrationBuilder.AlterColumn<string>(
name: "PrimaryConfiguration",
table: "DataConnections",
type: "nvarchar(4000)",
maxLength: 4000,
nullable: true,
oldClrType: typeof(string),
oldType: "nvarchar(max)",
oldNullable: true);
migrationBuilder.AlterColumn<string>(
name: "BackupConfiguration",
table: "DataConnections",
type: "nvarchar(4000)",
maxLength: 4000,
nullable: true,
oldClrType: typeof(string),
oldType: "nvarchar(max)",
oldNullable: true);
migrationBuilder.CreateIndex(
name: "IX_Instances_SiteId_UniqueName",
table: "Instances",
columns: new[] { "SiteId", "UniqueName" },
unique: true);
}
}
}
@@ -585,9 +585,11 @@ namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Migrations
b.HasIndex("AreaId");
b.HasIndex("SiteId");
b.HasIndex("TemplateId");
b.HasIndex("SiteId", "UniqueName")
b.HasIndex("UniqueName")
.IsUnique();
b.ToTable("Instances");
@@ -1335,8 +1337,7 @@ namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Migrations
SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property<int>("Id"));
b.Property<string>("BackupConfiguration")
.HasMaxLength(4000)
.HasColumnType("nvarchar(4000)");
.HasColumnType("nvarchar(max)");
b.Property<int>("FailoverRetryCount")
.ValueGeneratedOnAdd()
@@ -1349,8 +1350,7 @@ namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Migrations
.HasColumnType("nvarchar(200)");
b.Property<string>("PrimaryConfiguration")
.HasMaxLength(4000)
.HasColumnType("nvarchar(4000)");
.HasColumnType("nvarchar(max)");
b.Property<string>("Protocol")
.IsRequired()
@@ -943,10 +943,20 @@ VALUES
/// The two walkers must stay in step on window slack, depth cap and the DISTINCT-edges trick,
/// so there is exactly one copy of each.
/// </summary>
private static string ScopedAuditSource(IReadOnlyList<string>? sites) => sites is null
? "dbo.AuditLog" : "(SELECT * FROM dbo.AuditLog WHERE SourceSiteId IS NULL OR SourceSiteId IN (SELECT [value] FROM OPENJSON(@scopeSites)))";
private static void AddSiteScopeParameter(DbCommand command, IReadOnlyList<string>? sites)
{
if (sites is null) return;
var parameter = command.CreateParameter(); parameter.ParameterName = "@scopeSites";
parameter.Value = System.Text.Json.JsonSerializer.Serialize(sites); command.Parameters.Add(parameter);
}
private static async Task<ChainAnchor> ResolveChainAnchorAsync(
DbConnection conn,
Guid executionId,
CancellationToken ct)
CancellationToken ct, IReadOnlyList<string>? permittedSiteIds = null)
{
// --- Phase 1: walk up to the root ---------------------------------
// Climb ParentExecutionId until a node has no parent (root) or the
@@ -960,13 +970,15 @@ VALUES
await using (var upCmd = conn.CreateCommand())
{
upCmd.CommandText =
"SELECT TOP 1 ParentExecutionId FROM dbo.AuditLog " +
$"SELECT TOP 1 ParentExecutionId FROM {ScopedAuditSource(permittedSiteIds)} source " +
"WHERE ExecutionId = @cur AND ParentExecutionId IS NOT NULL;";
var pCur = upCmd.CreateParameter();
pCur.ParameterName = "@cur";
pCur.Value = rootExecutionId;
upCmd.Parameters.Add(pCur);
AddSiteScopeParameter(upCmd, permittedSiteIds);
var result = await upCmd.ExecuteScalarAsync(ct).ConfigureAwait(false);
parent = result is null or DBNull ? null : (Guid)result;
}
@@ -990,12 +1002,13 @@ VALUES
await using (var winCmd = conn.CreateCommand())
{
winCmd.CommandText =
"SELECT MIN(OccurredAtUtc) FROM dbo.AuditLog WHERE ExecutionId = @root;";
$"SELECT MIN(OccurredAtUtc) FROM {ScopedAuditSource(permittedSiteIds)} source WHERE ExecutionId = @root;";
var pWinRoot = winCmd.CreateParameter();
pWinRoot.ParameterName = "@root";
pWinRoot.Value = rootExecutionId;
winCmd.Parameters.Add(pWinRoot);
AddSiteScopeParameter(winCmd, permittedSiteIds);
var winResult = await winCmd.ExecuteScalarAsync(ct).ConfigureAwait(false);
rootFirstOccurred = winResult is null or DBNull ? null : (DateTime)winResult;
}
@@ -1026,10 +1039,10 @@ VALUES
/// ParentExecutionId of its child's edge. No separate
/// union-back CTE is needed.
/// </remarks>
private static string ChainCtePrefix(string edgesWindowPredicate) => $@"
private static string ChainCtePrefix(string edgesWindowPredicate, IReadOnlyList<string>? permittedSiteIds = null) => $@"
WITH Edges AS (
SELECT DISTINCT ExecutionId, ParentExecutionId
FROM dbo.AuditLog
FROM {ScopedAuditSource(permittedSiteIds)} scoped
WHERE ExecutionId IS NOT NULL
{edgesWindowPredicate}
),
@@ -1076,7 +1089,7 @@ VALUES
/// <inheritdoc />
public async Task<IReadOnlyList<ExecutionTreeNode>> GetExecutionTreeAsync(
Guid executionId,
CancellationToken ct = default)
CancellationToken ct = default, IReadOnlyList<string>? permittedSiteIds = null)
{
var conn = _context.Database.GetDbConnection();
var openedHere = false;
@@ -1089,7 +1102,7 @@ VALUES
try
{
// --- Phase 1 + window: shared with GetTraceEventsAsync (arch-review F9) ---
var anchor = await ResolveChainAnchorAsync(conn, executionId, ct).ConfigureAwait(false);
var anchor = await ResolveChainAnchorAsync(conn, executionId, ct, permittedSiteIds).ConfigureAwait(false);
// --- Phase 2: walk down from the root via a recursive CTE ---------
// The Edges/Chain prefix is ChainCtePrefix (see its remarks for the
@@ -1103,7 +1116,7 @@ VALUES
var nodes = new List<ExecutionTreeNode>();
await using (var downCmd = conn.CreateCommand())
{
downCmd.CommandText = ChainCtePrefix(anchor.EdgesWindowPredicate) + $@",
downCmd.CommandText = ChainCtePrefix(anchor.EdgesWindowPredicate, permittedSiteIds) + $@",
ChainIds AS (
SELECT ExecutionId FROM Chain
UNION
@@ -1125,21 +1138,22 @@ VALUES
MIN(a.ParentExecutionId) AS [ParentExecutionId],
COUNT(a.EventId) AS [RowCount],
(SELECT STRING_AGG(d.Category, ',')
FROM (SELECT DISTINCT a2.Category FROM dbo.AuditLog a2
FROM (SELECT DISTINCT a2.Category FROM {ScopedAuditSource(permittedSiteIds)} a2
WHERE a2.ExecutionId = ids.ExecutionId) d) AS [Channels],
(SELECT STRING_AGG(d.Status, ',')
FROM (SELECT DISTINCT a2.Status FROM dbo.AuditLog a2
FROM (SELECT DISTINCT a2.Status FROM {ScopedAuditSource(permittedSiteIds)} a2
WHERE a2.ExecutionId = ids.ExecutionId) d) AS [Statuses],
MIN(a.SourceSiteId) AS [SourceSiteId],
MIN(JSON_VALUE(a.DetailsJson,'$.sourceInstanceId')) AS [SourceInstanceId],
MIN(a.OccurredAtUtc) AS [FirstOccurredAtUtc],
MAX(a.OccurredAtUtc) AS [LastOccurredAtUtc]
FROM ChainIds ids
LEFT JOIN dbo.AuditLog a ON a.ExecutionId = ids.ExecutionId
LEFT JOIN {ScopedAuditSource(permittedSiteIds)} a ON a.ExecutionId = ids.ExecutionId
GROUP BY ids.ExecutionId
OPTION (MAXRECURSION {ExecutionChainMaxDepth});";
AddChainParameters(downCmd, anchor);
AddSiteScopeParameter(downCmd, permittedSiteIds);
await using var reader = await downCmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
while (await reader.ReadAsync(ct).ConfigureAwait(false))
@@ -1181,7 +1195,7 @@ VALUES
/// <inheritdoc />
public async Task<IReadOnlyList<AuditEvent>> GetTraceEventsAsync(
Guid executionId,
CancellationToken ct = default)
CancellationToken ct = default, IReadOnlyList<string>? permittedSiteIds = null)
{
// Flow Debugger Replay (design 2026-09-05 §7.1): same walk as GetExecutionTreeAsync
// — up to the root, then the Edges/Chain CTE down — but projecting the chain's
@@ -1197,24 +1211,28 @@ VALUES
}
var chainIds = new List<Guid>();
var rootId = executionId;
var chainTruncated = false;
try
{
var anchor = await ResolveChainAnchorAsync(conn, executionId, ct).ConfigureAwait(false);
var anchor = await ResolveChainAnchorAsync(conn, executionId, ct, permittedSiteIds).ConfigureAwait(false);
rootId = anchor.RootExecutionId;
await using (var idsCmd = conn.CreateCommand())
{
idsCmd.CommandText = ChainCtePrefix(anchor.EdgesWindowPredicate) + $@"
SELECT ExecutionId FROM Chain
idsCmd.CommandText = ChainCtePrefix(anchor.EdgesWindowPredicate, permittedSiteIds) + $@"
SELECT TOP (2049) ExecutionId FROM (SELECT ExecutionId FROM Chain
UNION
SELECT ParentExecutionId FROM Chain WHERE ParentExecutionId IS NOT NULL
SELECT ParentExecutionId FROM Chain WHERE ParentExecutionId IS NOT NULL) ids
OPTION (MAXRECURSION {ExecutionChainMaxDepth});";
AddChainParameters(idsCmd, anchor);
AddSiteScopeParameter(idsCmd, permittedSiteIds);
await using var reader = await idsCmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
while (await reader.ReadAsync(ct).ConfigureAwait(false))
{
if (!reader.IsDBNull(0)) chainIds.Add(reader.GetGuid(0));
if (!reader.IsDBNull(0)) { if (chainIds.Count == 2048) { chainTruncated = true; break; } chainIds.Add(reader.GetGuid(0)); }
}
}
}
@@ -1231,9 +1249,10 @@ VALUES
return Array.Empty<AuditEvent>();
}
var rows = await _context.Set<AuditLogRow>()
var query = _context.Set<AuditLogRow>()
.AsNoTracking()
.Where(r => r.ExecutionId != null && chainIds.Contains(r.ExecutionId.Value))
.Where(r => permittedSiteIds == null || r.SourceSiteId == null || permittedSiteIds.Contains(r.SourceSiteId))
// Outbound-calls plan (2026-09-06, D6): the replay set is the trace channel plus the
// three outbound channels (the script's API / DB / notification calls already sit in
// the tree under the run's execution id) plus the inbound row. SecuredWrite, Cluster
@@ -1245,11 +1264,49 @@ VALUES
|| r.Channel == AuditChannel.Notification
|| (r.Channel == AuditChannel.ApiInbound && r.Kind == AuditKind.InboundRequest))
.OrderBy(r => r.OccurredAtUtc)
.ThenBy(r => r.EventId)
.ToListAsync(ct)
.ConfigureAwait(false);
.ThenBy(r => r.EventId);
return rows.Select(RowToCanonical).ToList();
const int maxRows = 500;
const int maxBytes = 1024 * 1024 - 4096; // Reserve envelope/array metadata.
var events = new List<AuditEvent>();
var bytes = 0;
var truncated = chainTruncated;
// Small keyset pages also bound EF execution-strategy buffering. Reading
// all 501 candidate rows in one query could buffer hundreds of MiB before
// the consumer reaches its byte cutoff. Each page projects only details
// within the remaining budget; oversized events are omitted whole.
const int pageSize = 8;
DateTime? afterTime = null;
Guid afterId = default;
while (true)
{
var remaining = maxBytes - bytes;
var pageQuery = afterTime is { } cursor
? query.Where(r => r.OccurredAtUtc > cursor || (r.OccurredAtUtc == cursor && r.EventId.CompareTo(afterId) > 0))
: query;
var page = await pageQuery.OrderBy(r => r.OccurredAtUtc).ThenBy(r => r.EventId)
.Take(Math.Min(pageSize, maxRows + 1 - events.Count)).Select(r => new {
Row = new AuditLogRow {
EventId = r.EventId, OccurredAtUtc = r.OccurredAtUtc, Actor = r.Actor, Action = r.Action,
Outcome = r.Outcome, Channel = r.Channel, Target = r.Target, SourceNode = r.SourceNode,
CorrelationId = r.CorrelationId,
DetailsJson = r.DetailsJson != null && r.DetailsJson.Length <= remaining ? r.DetailsJson : null
}, DetailLength = r.DetailsJson == null ? 0 : r.DetailsJson.Length
}).ToListAsync(ct);
if (page.Count == 0) break;
foreach (var item in page)
{
if (item.DetailLength > remaining || events.Count >= maxRows)
return new TraceEventBatch(rootId, events, true);
var evt = RowToCanonical(item.Row);
var size = System.Text.Json.JsonSerializer.SerializeToUtf8Bytes(evt).Length;
if (size > maxBytes - bytes) return new TraceEventBatch(rootId, events, true);
events.Add(evt); bytes += size;
afterTime = item.Row.OccurredAtUtc; afterId = item.Row.EventId;
}
if (page.Count < pageSize) break;
}
return new TraceEventBatch(rootId, events, truncated);
}
/// <inheritdoc />
@@ -224,17 +224,37 @@ public class ScadaBridgeDbContext : DbContext, IDataProtectionKeyContext
public override int SaveChanges(bool acceptAllChangesOnSuccess)
{
GuardSecretWritesHaveAKeyRing();
return base.SaveChanges(acceptAllChangesOnSuccess);
try { return base.SaveChanges(acceptAllChangesOnSuccess); }
catch (DbUpdateException ex) when (IsInstanceNameConflict(ex))
{ throw DetachInstanceNameConflict(ex); }
}
/// <inheritdoc />
public override Task<int> SaveChangesAsync(
public override async Task<int> SaveChangesAsync(
bool acceptAllChangesOnSuccess, CancellationToken cancellationToken = default)
{
GuardSecretWritesHaveAKeyRing();
return base.SaveChangesAsync(acceptAllChangesOnSuccess, cancellationToken);
try { return await base.SaveChangesAsync(acceptAllChangesOnSuccess, cancellationToken); }
catch (DbUpdateException ex) when (IsInstanceNameConflict(ex))
{ throw DetachInstanceNameConflict(ex); }
}
private static InstanceNameConflictException DetachInstanceNameConflict(DbUpdateException error)
{
foreach (var entry in error.Entries) entry.State = EntityState.Detached;
return new InstanceNameConflictException(error);
}
private static bool IsInstanceNameConflict(DbUpdateException error)
=> error.InnerException is Microsoft.Data.SqlClient.SqlException sql
? sql.Number is 2601 or 2627 && sql.Message.Contains("IX_Instances_UniqueName", StringComparison.Ordinal)
: error.InnerException?.GetType().Name == "SqliteException"
&& error.InnerException.Message.Contains("UNIQUE constraint failed: Instances.UniqueName", StringComparison.Ordinal);
internal IDataProtector DataConnectionProtector => (_dataProtectionProvider
?? throw new InvalidOperationException("Data Protection is required to migrate protocol credentials."))
.CreateProtector(DataConnectionSecretConverter.Purpose);
/// <summary>True when this context was constructed with a real Data Protection provider.</summary>
internal bool HasSecretEncryptionProvider => _dataProtectionProvider is not null;
@@ -262,6 +282,8 @@ public class ScadaBridgeDbContext : DbContext, IDataProtectionKeyContext
SmsConfiguration => nameof(SmsConfiguration.AuthToken),
ExternalSystemDefinition => nameof(ExternalSystemDefinition.AuthConfiguration),
DatabaseConnectionDefinition => nameof(DatabaseConnectionDefinition.ConnectionString),
DataConnection connection when connection.PrimaryConfiguration is not null => nameof(DataConnection.PrimaryConfiguration),
DataConnection => nameof(DataConnection.BackupConfiguration),
_ => null
};
if (secretProperty is null)
@@ -280,14 +302,13 @@ public class ScadaBridgeDbContext : DbContext, IDataProtectionKeyContext
}
/// <summary>
/// Model cache key factory that folds <see cref="HasSecretEncryptionProvider"/> into the
/// cache key, so a write-capable (provider-bearing) context and a schema-only context do
/// not share a cached model.
/// Model cache key factory that folds the Data Protection provider identity into the
/// cache key, so different key rings and schema-only contexts never share converter instances.
/// </summary>
private sealed class SecretAwareModelCacheKeyFactory : IModelCacheKeyFactory
{
/// <summary>
/// Creates a model cache key that includes the Data Protection provider state.
/// Creates a model cache key that includes the Data Protection provider identity.
/// </summary>
/// <param name="context">The database context.</param>
/// <param name="designTime">Whether the model is being created at design-time.</param>
@@ -295,7 +316,7 @@ public class ScadaBridgeDbContext : DbContext, IDataProtectionKeyContext
public object Create(DbContext context, bool designTime)
=> (context.GetType(),
designTime,
(context as ScadaBridgeDbContext)?.HasSecretEncryptionProvider ?? false);
(context as ScadaBridgeDbContext)?._dataProtectionProvider);
/// <summary>
/// Creates a model cache key for run-time contexts.
@@ -350,6 +371,13 @@ public class ScadaBridgeDbContext : DbContext, IDataProtectionKeyContext
modelBuilder.Entity<DatabaseConnectionDefinition>()
.Property(d => d.ConnectionString)
.HasConversion(converter);
IDataProtector protocolProtector = _dataProtectionProvider is { } protocolProvider
? protocolProvider.CreateProtector(DataConnectionSecretConverter.Purpose)
: SchemaOnlyDataProtector.Instance;
var protocolConverter = new DataConnectionSecretConverter(protocolProtector);
modelBuilder.Entity<DataConnection>().Property(d => d.PrimaryConfiguration).HasConversion(protocolConverter);
modelBuilder.Entity<DataConnection>().Property(d => d.BackupConfiguration).HasConversion(protocolConverter);
}
/// <summary>
@@ -2899,6 +2899,87 @@ public class DataConnectionActor : UntypedActor, IWithTimers
// ── Native alarm subscriptions ──
private sealed class AlarmReplayState
{
public Dictionary<string, NativeAlarmTransition> Current { get; set; } = new();
public Dictionary<string, NativeAlarmTransition> Snapshot { get; } = new();
public Dictionary<string, NativeAlarmTransition> LiveDuringSnapshot { get; } = new();
public bool InSnapshot { get; set; }
public bool Ready { get; set; }
public Guid? Epoch { get; set; }
}
private readonly Dictionary<string, AlarmReplayState> _alarmReplay = new();
private void CacheAlarmTransition(string source, NativeAlarmTransition transition)
{
if (!_alarmReplay.TryGetValue(source, out var state))
_alarmReplay[source] = state = new AlarmReplayState();
switch (transition.Kind)
{
case AlarmTransitionKind.SnapshotStart:
if (transition.SnapshotId is not null && transition.SnapshotId == state.Epoch) return;
state.Snapshot.Clear();
state.LiveDuringSnapshot.Clear();
state.InSnapshot = true;
state.Epoch = transition.SnapshotId;
break;
case AlarmTransitionKind.Snapshot:
if (transition.SnapshotId != state.Epoch || (transition.SnapshotId is not null && !state.InSnapshot)) return;
state.InSnapshot = true;
state.Snapshot[transition.SourceReference] = transition;
break;
case AlarmTransitionKind.SnapshotComplete:
if (transition.SnapshotId != state.Epoch || (transition.SnapshotId is not null && !state.InSnapshot)) return;
state.Current = SnapshotWithLiveChanges(state);
state.Snapshot.Clear();
state.LiveDuringSnapshot.Clear();
state.InSnapshot = false;
state.Ready = true;
break;
default:
if (state.LiveDuringSnapshot.TryGetValue(transition.SourceReference, out var recent)
&& transition.TransitionTime < recent.TransitionTime) return;
if (state.Current.TryGetValue(transition.SourceReference, out var prior)
&& transition.TransitionTime < prior.TransitionTime) return;
if (state.InSnapshot) state.LiveDuringSnapshot[transition.SourceReference] = transition;
if (!transition.Condition.Active && transition.Condition.Acknowledged)
state.Current.Remove(transition.SourceReference);
else
state.Current[transition.SourceReference] = transition;
break;
}
}
private static Dictionary<string, NativeAlarmTransition> SnapshotWithLiveChanges(AlarmReplayState state)
{
var result = new Dictionary<string, NativeAlarmTransition>(state.Snapshot);
foreach (var (key, live) in state.LiveDuringSnapshot)
{
if (result.TryGetValue(key, out var snapshot) && snapshot.TransitionTime > live.TransitionTime) continue;
if (!live.Condition.Active && live.Condition.Acknowledged) result.Remove(key);
else result[key] = live;
}
return result;
}
private void ReplayAlarmsTo(string source, IActorRef subscriber)
{
if (!_alarmReplay.TryGetValue(source, out var state) || (!state.Ready && !state.InSnapshot)) return;
var epoch = state.InSnapshot ? state.Epoch : Guid.NewGuid();
var boundary = new NativeAlarmTransition("", source, "", AlarmTransitionKind.SnapshotStart,
new AlarmConditionState(false, true, null, AlarmShelveState.Unshelved, false, 0),
"", "", "", "", "", null, DateTimeOffset.UtcNow, "", "", SnapshotSourceReference: source, SnapshotId: epoch);
// Legacy Mx snapshots have no start or epoch. Mid-frame subscribers get
// the prefix already received and then share the remaining live frame.
if (epoch is not null)
subscriber.Tell(new NativeAlarmTransitionUpdate(_connectionName, boundary));
foreach (var transition in (state.InSnapshot ? SnapshotWithLiveChanges(state) : state.Current).Values)
subscriber.Tell(new NativeAlarmTransitionUpdate(_connectionName, transition with
{ Kind = AlarmTransitionKind.Snapshot, SnapshotSourceReference = source, SnapshotId = epoch }));
if (!state.InSnapshot)
subscriber.Tell(new NativeAlarmTransitionUpdate(_connectionName, boundary with { Kind = AlarmTransitionKind.SnapshotComplete }));
}
private void HandleSubscribeAlarms(SubscribeAlarmsRequest request)
{
var subscriber = Sender;
@@ -2920,7 +3001,7 @@ public class DataConnectionActor : UntypedActor, IWithTimers
_alarmSourceSubscribers[request.SourceReference] = subs;
IndexAlarmSource(request.SourceReference);
}
subs.Add(subscriber);
var newlyAdded = subs.Add(subscriber);
Context.Watch(subscriber);
// S10: the adapter feed carries a single shared condition filter per source.
// A second subscriber with a different filter silently overwrites the first
@@ -2943,6 +3024,7 @@ public class DataConnectionActor : UntypedActor, IWithTimers
if (_alarmSubscriptionIds.ContainsKey(request.SourceReference) ||
_alarmSubscribesInFlight.Contains(request.SourceReference))
{
if (newlyAdded) ReplayAlarmsTo(request.SourceReference, subscriber);
subscriber.Tell(new SubscribeAlarmsResponse(
request.CorrelationId, request.InstanceUniqueName, true, null, now));
return;
@@ -3081,12 +3163,15 @@ public class DataConnectionActor : UntypedActor, IWithTimers
var scope = transition.SnapshotSourceReference
?? (string.IsNullOrEmpty(transition.SourceObjectReference) ? null : transition.SourceObjectReference);
var targets = scope is null
? _alarmSourceSubscribers.Values
: _alarmSourceSubscribers.Where(kv => kv.Key == scope).Select(kv => kv.Value);
foreach (var subs in targets)
foreach (var sub in subs)
? _alarmSourceSubscribers
: _alarmSourceSubscribers.Where(kv => kv.Key == scope);
foreach (var (source, subscribers) in targets)
{
CacheAlarmTransition(source, transition);
foreach (var sub in subscribers)
if (notified.Add(sub))
sub.Tell(new NativeAlarmTransitionUpdate(_connectionName, transition));
}
return;
}
@@ -3110,6 +3195,7 @@ public class DataConnectionActor : UntypedActor, IWithTimers
!predicate.IsAllowed(transition))
continue;
CacheAlarmTransition(sourceRef, transition);
foreach (var sub in subs)
{
if (notified.Add(sub))
@@ -3215,6 +3301,7 @@ public class DataConnectionActor : UntypedActor, IWithTimers
// No subscribers remain for this source — tear down the adapter feed.
_alarmSourceSubscribers.Remove(sourceReference);
_alarmReplay.Remove(sourceReference);
UnindexAlarmSource(sourceReference);
_alarmSourceFilter.Remove(sourceReference);
_alarmSourceFilterPredicate.Remove(sourceReference);
@@ -3237,6 +3324,7 @@ public class DataConnectionActor : UntypedActor, IWithTimers
return;
_alarmSubscriptionIds.Clear();
_alarmReplay.Clear();
_alarmSubscribesInFlight.Clear();
var self = Self;
var generation = _adapterGeneration;
@@ -319,8 +319,8 @@ public class MxGatewayDataConnection
// MxGateway has no server-side CONDITION filter either, so conditionFilter is
// still not forwarded: the DataConnectionActor's per-source + per-condition-type
// gate stays authoritative, uniform with the OPC UA WhereClause stance.
if (_alarmCts != null && AlarmFilterPrefix.Covers(_alarmStreamPrefix, sourceReference))
return Task.FromResult(subscriptionId);
// Even a covered new source needs a standing-state snapshot. Reopen
// the shared stream; DCL separately replays same-source co-subscribers.
// A new source outside the live prefix (or the first subscriber): (re)open the
// stream on the recomputed union. The source replays a fresh snapshot, which
@@ -336,7 +336,23 @@ public class MxGatewayDataConnection
var token = _alarmCts.Token;
var client = _client!;
var prefix = _alarmStreamPrefix.Length == 0 ? null : _alarmStreamPrefix;
_ = Task.Run(() => client.RunAlarmStreamAsync(prefix, t => callback(t), token), token);
var epoch = Guid.NewGuid();
void Forward(Commons.Types.Alarms.NativeAlarmTransition transition)
{
lock (_alarmLock)
{
if (token.IsCancellationRequested) return;
if (transition.Kind == AlarmTransitionKind.SnapshotStart) epoch = Guid.NewGuid();
callback(transition.Kind is AlarmTransitionKind.SnapshotStart or AlarmTransitionKind.Snapshot or AlarmTransitionKind.SnapshotComplete
? transition with { SnapshotId = epoch } : transition);
}
}
_ = Task.Run(async () =>
{
Forward(MxGatewayAlarmMapper.SnapshotComplete() with { Kind = AlarmTransitionKind.SnapshotStart });
if (token.IsCancellationRequested) return;
await client.RunAlarmStreamAsync(prefix, Forward, token);
}, token);
}
return Task.FromResult(subscriptionId);
@@ -427,6 +427,9 @@ public sealed class RealMxGatewayClient : IMxGatewayClient
AlarmFilterPrefix = alarmFilterPrefix ?? string.Empty,
};
// Every transport reopen produces a fresh authoritative frame,
// including an empty snapshot after the previous stream faulted.
onTransition(MxGatewayAlarmMapper.SnapshotComplete() with { Kind = Commons.Types.Enums.AlarmTransitionKind.SnapshotStart });
await foreach (var message in _client!.StreamAlarmsAsync(request, ct)
.WithCancellation(ct).ConfigureAwait(false))
{
@@ -96,6 +96,22 @@ public class RealOpcUaClient : IOpcUaClient
/// <summary>Raised when the OPC UA connection is lost.</summary>
public event Action? ConnectionLost;
/// <summary>Selects exactly the requested protection mode; protected discovery failures fail closed.</summary>
internal static EndpointDescription DiscoverEndpoint(string url, MessageSecurityMode requested,
Func<EndpointDescriptionCollection> discover)
{
EndpointDescriptionCollection endpoints;
try { endpoints = discover(); }
catch (Exception ex)
{
if (requested != MessageSecurityMode.None)
throw new InvalidOperationException($"OPC UA endpoint discovery failed; requested security mode {requested} cannot be verified.", ex);
return new EndpointDescription(url);
}
return endpoints.FirstOrDefault(e => e.SecurityMode == requested)
?? throw new InvalidOperationException($"OPC UA requested security mode {requested} is unavailable. Available modes: {string.Join(", ", endpoints.Select(e => e.SecurityMode).Distinct())}.");
}
/// <inheritdoc />
public async Task ConnectAsync(string endpointUrl, OpcUaConnectionOptions? options = null, CancellationToken cancellationToken = default)
{
@@ -140,24 +156,13 @@ public class RealOpcUaClient : IOpcUaClient
}
// Discover endpoints from the server, pick the preferred security mode
EndpointDescription? endpoint;
try
var endpoint = DiscoverEndpoint(endpointUrl, preferredSecurityMode, () =>
{
#pragma warning disable CS0618
using var discoveryClient = DiscoveryClient.Create(new Uri(endpointUrl));
return discoveryClient.GetEndpoints(null);
#pragma warning restore CS0618
#pragma warning disable CS0618
var endpoints = discoveryClient.GetEndpoints(null);
#pragma warning restore CS0618
endpoint = endpoints
.Where(e => e.SecurityMode == preferredSecurityMode)
.FirstOrDefault() ?? endpoints.FirstOrDefault();
}
catch
{
// Fallback: construct endpoint description manually
endpoint = new EndpointDescription(endpointUrl);
}
});
// The server advertises its own base address, which is frequently unroutable from here
// (a 0.0.0.0 wildcard bind, or an internal container/NAT hostname). Swap it back to the
@@ -377,21 +382,13 @@ public class RealOpcUaClient : IOpcUaClient
await appConfig.ValidateAsync(ApplicationType.Client);
// Discover endpoints, pick the preferred security mode (same logic as ConnectAsync).
EndpointDescription? endpoint;
try
var endpoint = DiscoverEndpoint(endpointUrl, preferredSecurityMode, () =>
{
#pragma warning disable CS0618
using var discoveryClient = DiscoveryClient.Create(new Uri(endpointUrl));
var endpoints = discoveryClient.GetEndpoints(null);
return discoveryClient.GetEndpoints(null);
#pragma warning restore CS0618
endpoint = endpoints
.Where(ep => ep.SecurityMode == preferredSecurityMode)
.FirstOrDefault() ?? endpoints.FirstOrDefault();
}
catch
{
endpoint = new EndpointDescription(endpointUrl);
}
});
// Same reachability rewrite as ConnectAsync — a probe against a server advertising a
// 0.0.0.0 / NAT base address must dial the reachable host, not the advertised one.
@@ -6,7 +6,9 @@ namespace ZB.MOM.WW.ScadaBridge.DelmiaNotifier;
/// <summary>
/// <see cref="IRecipeSender"/> over a real <see cref="HttpClient"/>: POSTs the recipe payload to
/// <c>{baseUrl}/api/DelmiaRecipeDownload</c> with the <c>X-API-Key</c> header. A reachable server
/// (any status) is <see cref="AttemptKind.Connected"/>; a connection/timeout failure is <see cref="AttemptKind.ConnectFailed"/>.
/// response is final except for the explicit pre-execution standby marker. Only proven
/// connection refusal/name resolution failures permit another URL; timeouts and other
/// transport failures have an uncertain application outcome.
/// </summary>
internal sealed class HttpRecipeSender(HttpClient http, string apiKey) : IRecipeSender
{
@@ -25,6 +27,9 @@ internal sealed class HttpRecipeSender(HttpClient http, string apiKey) : IRecipe
using var response = await http.SendAsync(request, ct);
var status = (int)response.StatusCode;
if (status == 503 && response.Headers.TryGetValues("X-ScadaBridge-Standby", out var marker)
&& marker.SequenceEqual(new[] { "1" }))
return new AttemptOutcome(AttemptKind.StandbyRejected, status, null, "standby rejected before execution");
RecipeDownloadResult? body = null;
if (response.IsSuccessStatusCode)
{
@@ -36,12 +41,14 @@ internal sealed class HttpRecipeSender(HttpClient http, string apiKey) : IRecipe
}
catch (HttpRequestException ex)
{
return new AttemptOutcome(AttemptKind.ConnectFailed, 0, null, ex.Message);
var knownNotSent = ex.HttpRequestError == HttpRequestError.NameResolutionError
|| ex.InnerException is System.Net.Sockets.SocketException { SocketErrorCode: System.Net.Sockets.SocketError.ConnectionRefused };
return new AttemptOutcome(knownNotSent ? AttemptKind.ConnectFailed : AttemptKind.OutcomeUnknown, 0, null, ex.Message);
}
catch (OperationCanceledException ex) when (!ct.IsCancellationRequested)
{
// No external cancellation requested → this is an HttpClient.Timeout, i.e. the node never answered.
return new AttemptOutcome(AttemptKind.ConnectFailed, 0, null, "timeout: " + ex.Message);
// A missing response does not prove the node failed to apply the request.
return new AttemptOutcome(AttemptKind.OutcomeUnknown, 0, null, "timeout: " + ex.Message);
}
}
@@ -5,6 +5,8 @@ internal enum AttemptKind
{
Connected,
ConnectFailed,
StandbyRejected,
OutcomeUnknown,
}
/// <summary>Result of one POST attempt against a single base URL.</summary>
@@ -4,14 +4,13 @@ namespace ZB.MOM.WW.ScadaBridge.DelmiaNotifier;
internal sealed record NotifyResult(bool Ok, string Reason);
/// <summary>
/// Connect-failure-only failover loop: tries each base URL in order. A node that responds at all is
/// authoritative — its answer is final (success, business rejection, or HTTP error alike). Only a
/// failure to connect rolls over to the next URL; if every URL fails to connect the last error is reported.
/// Tries another URL only after proven pre-execution connection failure or explicit standby
/// rejection. Business responses are final; ambiguous response loss reports OUTCOME_UNKNOWN.
/// </summary>
internal static class Notifier
{
/// <summary>
/// Runs the connect-failure-only failover loop across <paramref name="baseUrls"/>,
/// Runs the pre-execution-only failover loop across <paramref name="baseUrls"/>,
/// sending <paramref name="payload"/> via <paramref name="sender"/> until a node
/// responds (connects) or every URL fails to connect.
/// </summary>
@@ -29,12 +28,15 @@ internal static class Notifier
{
var outcome = await sender.SendAsync(baseUrl, payload, ct);
if (outcome.Kind == AttemptKind.ConnectFailed)
if (outcome.Kind is AttemptKind.ConnectFailed or AttemptKind.StandbyRejected)
{
lastError = outcome.Error ?? "connection failed";
continue; // unreachable node → try the next
}
if (outcome.Kind == AttemptKind.OutcomeUnknown)
return new NotifyResult(false, "OUTCOME_UNKNOWN: the request may have been applied; verify before retrying.");
// Connected — this node's answer is authoritative; never fail over past it.
if (IsSuccessStatus(outcome.StatusCode))
{
@@ -54,12 +54,11 @@ WWNotifier.exe -m Z28061 -d "C:\recipes\job.nc" -w WO12345 -p PN-7788 -s 0100 -u
## Delivery semantics
The `BaseUrls` failover loop is at-least-once, not exactly-once: a connection reset *after* the
target already processed the POST, or a slow response that exceeds `TimeoutSeconds`, both look like
an unreachable node to this client and cause a re-POST to the next URL — even though the first
attempt may already have succeeded. The deployed `DelmiaRecipeDownload` inbound method must be
idempotent on `(MachineCode, DownloadPath, WorkOrderNumber)` so a duplicate delivery is harmless;
see the design doc's "Delivery semantics: at-least-once" section for details.
The notifier retries only when a request is known not to have executed: connection refusal,
name-resolution failure, or the server's explicit standby rejection. A connection reset after
sending or a response timeout returns `OUTCOME_UNKNOWN` without replay. Operators must verify
an uncertain result before manually retrying. Idempotent method implementations remain useful
because callers can deliberately retry, but the notifier does not blindly repeat ambiguous sends.
## Configuration — `appsettings.json`
@@ -77,10 +76,9 @@ Placed next to the exe (copied to the output on build/publish):
```
- **`BaseUrls`** — comma-separated failover list of **base URLs only**; the app appends
`/api/DelmiaRecipeDownload`. URLs are tried in order with **connect-failure-only failover**: a node
that responds at all (even `4xx`/`5xx`) is authoritative and its answer is final; only a node that
fails to connect (refused/reset, DNS, TLS, timeout) rolls over to the next. If every URL fails to
connect → `NO` + last connection error.
`/api/DelmiaRecipeDownload`. URLs are tried in order after proven DNS/connection refusal or
explicit standby rejection. Other HTTP responses are final; ambiguous transport errors return
`NO` + `OUTCOME_UNKNOWN` without trying another URL.
- **`TimeoutSeconds`** — per-attempt HTTP timeout (default `30`).
- **`LogPath`** — optional diagnostic log file, relative to the exe. Omit to log to stderr only.
@@ -150,7 +148,13 @@ bash docker/verify-delmia-notifier.sh # runs the managed build from a scratc
| A | active node first in `BaseUrls` | `YES` |
| B | standby first (standby relays to the active) | `YES` |
| C | bad key | `NO` + `HTTP 401` (an answer is final — no failover) |
| D | active `docker kill`ed, attempt inside the takeover window | `NO` + `HTTP 503` — expected: connect-failure-only failover, the survivor is not yet active |
| D | active `docker kill`ed, attempt inside the takeover window | historical baseline: `NO` + `HTTP 503`; the updated client retries explicit standby refusals and still fails if no node is active |
| E | survivor `/health/active` = 200 (~27 s) | `YES` (first URL refused → second URL) |
| F | victim restarted and rejoined as standby | `YES` (victim relays) |
| G | `DownloadPath` ending in `.bad` | `NO` + `Simulated reject: bad path or missing work order` |
### Safe failover
The notifier advances to the next configured URL after a proven connection refusal/name-resolution failure, or HTTP 503 carrying the server's exact `X-ScadaBridge-Standby: 1` pre-execution marker. A generic 503 or business rejection is final. Both centrals may therefore be listed even when their internal standby-forwarding peer lists are absent.
A timeout or other response loss is reported as `OUTCOME_UNKNOWN` with a failed result and no automatic replay: the first node may already have applied the download. Verify the operation before manually retrying. Configure trusted central endpoints; the standby marker is a server protocol contract, not permission to retry arbitrary HTTP errors.
@@ -568,6 +568,7 @@ public class DeploymentService
"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);
@@ -1065,7 +1066,11 @@ public class DeploymentService
// CancellationToken.None inside (see MarkDeploymentFailedAsync) so the
// already-cancelled token cannot abort the cleanup writes themselves.
await ReleasePreparedAsync(prepared, finalizedUpTo, user, ex);
throw;
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(
@@ -1149,11 +1154,9 @@ public class DeploymentService
/// <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. That is bounded by the probe's watermark
/// memo (unchanged graph ⇒ no flatten at all on the second and later calls, and
/// the Deployments page's staleness sweep warms the same memo), and it buys
/// back far more than it costs on the common mostly-unchanged site, which now
/// sends no site round-trips at all instead of one per instance.
/// 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>
@@ -1,147 +1,30 @@
using System.Collections.Concurrent;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport;
namespace ZB.MOM.WW.ScadaBridge.DeploymentManager;
/// <summary>
/// <see cref="IStaleInstanceProbe"/> backed by the deployment flattening pipeline.
/// Re-flattens an instance through <see cref="IFlatteningPipeline"/> and returns
/// the computed revision hash — the exact same value
/// <c>DeploymentService.GetDeploymentComparisonAsync</c> compares against the
/// deployed snapshot to decide staleness. Hosted in DeploymentManager so the
/// Transport component (which references only Commons + TemplateEngine) can probe
/// staleness without taking a DeploymentManager project reference.
///
/// <para>
/// <b>Watermark fast path.</b> A bundle import probes EVERY instance derived from
/// an overwritten template, and the Deployments page sweeps the fleet — both pay a
/// full flatten per instance for an answer that is usually "unchanged". A probe
/// result is therefore memoised against the
/// <see cref="ITemplateGraphWatermark"/> reading that produced it: the instance's
/// own version, the graph's structure version (which covers chain MEMBERSHIP), and
/// the version of every template the flatten actually walked. When all of those are
/// unchanged the recorded hash is still current by construction and the flatten is
/// skipped.
/// </para>
///
/// <para>
/// The fast path can only ever cause EXTRA work, never stale work: the watermark is
/// in-memory and process-local, so a restart, a failover to the other central node,
/// or any mutation at all simply misses and falls back to the authoritative
/// flatten.
/// </para>
/// </summary>
/// <summary>Recomputes deployment revisions from authoritative configuration on every probe.
/// Process-local watermarks cannot prove freshness after another central changes the shared store.</summary>
public sealed class StaleInstanceProbe : IStaleInstanceProbe
{
/// <summary>
/// Ceiling on memoised probe results. Bounded because the key space is the
/// instance table and this cache is process-wide and static; on overflow it is
/// dropped wholesale, which is safe here (unlike the script-compile verdict
/// cache) because a miss costs one flatten and leaks nothing.
/// </summary>
private const int MaxCachedProbes = 4096;
private static readonly ConcurrentDictionary<int, ProbeMemo> Memos = new();
private readonly IFlatteningPipeline _flatteningPipeline;
private readonly ITemplateGraphWatermark _watermark;
/// <summary>Initializes a new <see cref="StaleInstanceProbe"/>.</summary>
/// <param name="flatteningPipeline">The deployment flattening pipeline used to recompute the current revision hash.</param>
/// <param name="watermark">Graph version watermark backing the fast path.</param>
/// <summary>Initializes the authoritative probe. Watermark remains a constructor compatibility dependency.</summary>
public StaleInstanceProbe(IFlatteningPipeline flatteningPipeline, ITemplateGraphWatermark watermark)
{
_flatteningPipeline = flatteningPipeline ?? throw new ArgumentNullException(nameof(flatteningPipeline));
_watermark = watermark ?? throw new ArgumentNullException(nameof(watermark));
ArgumentNullException.ThrowIfNull(watermark);
}
/// <summary>
/// A previously computed revision hash together with the exact watermark
/// readings it was computed under.
/// </summary>
/// <param name="Structure">Graph structure version at flatten time.</param>
/// <param name="InstanceVersion">The instance's own version at flatten time.</param>
/// <param name="TemplateVersions">Version every template the flatten walked was READ at, by template id.</param>
/// <param name="RevisionHash">The hash that reading produced.</param>
private sealed record ProbeMemo(
long Structure,
long InstanceVersion,
IReadOnlyDictionary<int, long> TemplateVersions,
string RevisionHash);
/// <inheritdoc />
public async Task<string?> GetCurrentRevisionHashAsync(int instanceId, CancellationToken cancellationToken = default)
{
if (TryGetMemoisedHash(instanceId, out var memoised))
return memoised;
// The pipeline returns a Result; a flatten failure (e.g. unresolvable
// template chain mid-import) yields null so the caller treats the
// instance as "staleness indeterminate" and skips it. Flattening reuses
// the scoped ITemplateEngineRepository, so it observes template rows the
// in-flight import has staged on the shared change tracker.
// Staleness only needs the revision hash — skip the expensive Roslyn
// script-compilation stage (this probe is called per-instance across a whole
// bundle import in BundleImporter.ComputeStaleInstanceIdsAsync).
//
// Watermark readings are taken BEFORE the flatten they describe. A mutation
// that lands during the flatten therefore bumps a version past the recorded
// one and invalidates the memo on the next read, rather than being captured
// as if it had already been included. The structure and instance versions
// are read here; the per-template versions come from the session, which
// records each template's version at the moment it read that template (see
// FlattenSession.VisitedTemplateVersions) — re-reading them after the
// flatten would pair a pre-edit hash with a post-edit version and make the
// memo serve "not stale" indefinitely.
var structure = _watermark.StructureVersion;
var instanceVersion = _watermark.GetInstanceVersion(instanceId);
var session = _flatteningPipeline.CreateSession();
var result = await _flatteningPipeline
.FlattenAndValidateAsync(instanceId, cancellationToken, validateScripts: false, session)
.ConfigureAwait(false);
if (result.IsFailure)
return null;
if (Memos.Count >= MaxCachedProbes)
Memos.Clear();
Memos[instanceId] = new ProbeMemo(
structure, instanceVersion, session.VisitedTemplateVersions, result.Value.RevisionHash);
return result.Value.RevisionHash;
var result = await _flatteningPipeline.FlattenAndValidateAsync(instanceId, cancellationToken,
validateScripts: false, session).ConfigureAwait(false);
return result.IsSuccess ? result.Value.RevisionHash : null;
}
/// <summary>
/// Returns the memoised revision hash when every watermark reading it was
/// computed under is still current.
/// </summary>
private bool TryGetMemoisedHash(int instanceId, out string? hash)
{
hash = null;
if (!Memos.TryGetValue(instanceId, out var memo))
return false;
if (memo.Structure != _watermark.StructureVersion)
return false;
if (memo.InstanceVersion != _watermark.GetInstanceVersion(instanceId))
return false;
foreach (var (templateId, version) in memo.TemplateVersions)
{
if (_watermark.GetTemplateVersion(templateId) != version)
return false;
}
hash = memo.RevisionHash;
return true;
}
/// <summary>
/// Drops every memoised probe result. Test hook — production invalidation is
/// driven entirely by the watermark.
/// </summary>
public static void ClearMemos() => Memos.Clear();
/// <summary>Compatibility hook: probes no longer retain process-global results.</summary>
public static void ClearMemos() { }
}
@@ -16,7 +16,7 @@ namespace ZB.MOM.WW.ScadaBridge.ExternalSystemGateway;
/// Database.Connection("name") — returns ADO.NET SqlConnection (connection pooling).
/// Database.CachedWrite("name", "sql", params) — submits to S&amp;F engine.
/// </summary>
public class DatabaseGateway : IDatabaseGateway
public class DatabaseGateway : IDatabaseGateway, ISandboxDatabaseConnectionSource
{
private readonly IExternalSystemRepository _repository;
private readonly StoreAndForwardService? _storeAndForward;
@@ -38,6 +38,14 @@ public class DatabaseGateway : IDatabaseGateway
_storeAndForward = storeAndForward;
}
/// <inheritdoc />
public async Task<string> GetSandboxConnectionStringAsync(string connectionName, CancellationToken cancellationToken)
{
var definition = await ResolveConnectionAsync(connectionName, cancellationToken)
?? throw new InvalidOperationException($"Database connection '{connectionName}' not found");
return definition.ConnectionString;
}
/// <inheritdoc />
public async Task<DbConnection> GetConnectionAsync(
string connectionName,
@@ -0,0 +1,25 @@
using Grpc.AspNetCore.Server;
namespace ZB.MOM.WW.ScadaBridge.Host;
/// <summary>Assigns endpoints to actual local sockets rather than caller-controlled authority headers.</summary>
public static class CentralListenerIsolation
{
/// <summary>Runs after routing and before authentication or endpoint dispatch.</summary>
public static IApplicationBuilder UseCentralListenerIsolation(this IApplicationBuilder app,
int grpcPort, int? metricsPort, IReadOnlyCollection<int> httpPorts)
=> app.Use(async (context, next) =>
{
var port = context.Connection.LocalPort;
var grpc = context.GetEndpoint()?.Metadata.GetMetadata<GrpcMethodMetadata>() is not null;
var dedicatedMetrics = metricsPort == port && port != grpcPort && !httpPorts.Contains(port);
var metricsOrHealth = context.Request.Path == "/metrics" || context.Request.Path == "/healthz"
|| context.Request.Path == "/health/ready" || context.Request.Path == "/health/active";
if ((port == grpcPort) != grpc || (dedicatedMetrics && !metricsOrHealth))
{
context.Response.StatusCode = StatusCodes.Status404NotFound;
return;
}
await next(context);
});
}
@@ -678,6 +678,7 @@ try
app.UseWebSockets();
app.UseRouting();
app.UseCentralListenerIsolation(centralGrpcPort, centralMetricsPort, httpPorts);
app.UseAuthentication();
app.UseAuthorization();
app.UseAntiforgery();
@@ -75,4 +75,5 @@
<ProjectReference Include="../ZB.MOM.WW.ScadaBridge.ManagementService/ZB.MOM.WW.ScadaBridge.ManagementService.csproj" />
</ItemGroup>
<ItemGroup><ProjectReference Include="../ZB.MOM.WW.ScadaBridge.CentralUI.SandboxWorker/ZB.MOM.WW.ScadaBridge.CentralUI.SandboxWorker.csproj" /></ItemGroup>
</Project>
@@ -17,9 +17,10 @@ namespace ZB.MOM.WW.ScadaBridge.InboundAPI;
/// first response that is not itself a standby refusal. A peer that is also a standby
/// (mid-failover, or a misconfigured list that includes this node) answers 503 with the
/// <see cref="StandbyResponseHeader"/> marker, which the forwarder treats like a
/// connection failure: try the next peer. Connection errors and per-attempt timeouts
/// (<see cref="InboundApiOptions.StandbyForwardTimeout"/>) also move to the next peer.
/// When every peer fails the caller gets 503 <c>ACTIVE_NODE_UNREACHABLE</c>.</para>
/// known pre-execution rejection: try the next peer. Only proven name-resolution or
/// connection-refused failures also permit retry. A timeout or ambiguous transport failure
/// returns 502 OUTCOME_UNKNOWN without replay. One budget spans every attempt. When all
/// permitted attempts fail before execution, returns 503 ACTIVE_NODE_UNREACHABLE.</para>
///
/// <para><b>Loop guard.</b> Every relayed request carries
/// <see cref="ForwardedRequestHeader"/>. <see cref="InboundApiEndpointFilter"/> never
@@ -94,33 +95,31 @@ public sealed class HttpStandbyRequestForwarder : IStandbyRequestForwarder
// Buffer the body once so it can be replayed to a second peer. The filter has
// already enforced MaxRequestBodyBytes (Content-Length up front, and the
// per-request Kestrel cap for chunked bodies), so this is bounded.
byte[] body;
using (var buffer = new MemoryStream())
{
await context.Request.Body.CopyToAsync(buffer, cancellationToken).ConfigureAwait(false);
body = buffer.ToArray();
}
var body = await RelayProvenance.ReadBoundedAsync(context.Request.Body, opts.MaxRequestBodyBytes, cancellationToken);
var provenance = RelayProvenance.Sign(context, body, opts.ApiKeyPepper);
var client = _httpClientFactory.CreateClient(HttpClientName);
var pathAndQuery = context.Request.Path.Value + context.Request.QueryString.Value;
var forwardedBy = string.IsNullOrEmpty(context.Request.Host.Host) ? "1" : context.Request.Host.Host;
// A single caller-visible budget spans all known-preexecution attempts.
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
deadline.CancelAfter(opts.StandbyForwardTimeout);
foreach (var peer in peers)
{
cancellationToken.ThrowIfCancellationRequested();
using var attemptCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
attemptCts.CancelAfter(opts.StandbyForwardTimeout);
if (deadline.IsCancellationRequested) return Unreachable();
try
{
using var request = BuildRequest(context, peer, pathAndQuery, body, forwardedBy);
var response = await client
.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, attemptCts.Token)
using var request = BuildRequest(context, peer, pathAndQuery, body, forwardedBy, provenance);
using var response = await client
.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, deadline.Token)
.ConfigureAwait(false);
if (response.StatusCode == HttpStatusCode.ServiceUnavailable
&& response.Headers.Contains(StandbyResponseHeader))
&& response.Headers.TryGetValues(StandbyResponseHeader, out var marker) && marker.SequenceEqual(new[] { "1" }))
{
response.Dispose();
_logger.LogInformation(
@@ -129,7 +128,7 @@ public sealed class HttpStandbyRequestForwarder : IStandbyRequestForwarder
}
var responseBody = await response.Content
.ReadAsByteArrayAsync(attemptCts.Token)
.ReadAsByteArrayAsync(deadline.Token)
.ConfigureAwait(false);
var contentType = response.Content.Headers.ContentType?.ToString();
var status = (int)response.StatusCode;
@@ -146,14 +145,17 @@ public sealed class HttpStandbyRequestForwarder : IStandbyRequestForwarder
}
catch (OperationCanceledException)
{
_logger.LogWarning(
"Inbound API relay: peer {Peer} did not answer within {Timeout} — trying the next peer",
peer, opts.StandbyForwardTimeout);
_logger.LogWarning("Inbound relay deadline expired after dispatch to {Peer}; outcome uncertain", peer);
return OutcomeUnknown();
}
catch (HttpRequestException ex) when (KnownNotDispatched(ex))
{
_logger.LogWarning(ex, "Inbound relay could not connect to {Peer}; trying next peer", peer);
}
catch (HttpRequestException ex)
{
_logger.LogWarning(ex,
"Inbound API relay: peer {Peer} unreachable — trying the next peer", peer);
_logger.LogWarning(ex, "Inbound relay lost response from {Peer}; outcome uncertain", peer);
return OutcomeUnknown();
}
}
@@ -164,7 +166,7 @@ public sealed class HttpStandbyRequestForwarder : IStandbyRequestForwarder
}
private static HttpRequestMessage BuildRequest(
HttpContext context, string peer, string pathAndQuery, byte[] body, string forwardedBy)
HttpContext context, string peer, string pathAndQuery, byte[] body, string forwardedBy, string? provenance)
{
var request = new HttpRequestMessage(
new HttpMethod(context.Request.Method),
@@ -174,7 +176,9 @@ public sealed class HttpStandbyRequestForwarder : IStandbyRequestForwarder
foreach (var header in context.Request.Headers)
{
if (HopByHopRequestHeaders.Contains(header.Key)
|| string.Equals(header.Key, ForwardedRequestHeader, StringComparison.OrdinalIgnoreCase))
|| string.Equals(header.Key, ForwardedRequestHeader, StringComparison.OrdinalIgnoreCase)
|| string.Equals(header.Key, RelayProvenance.Header, StringComparison.OrdinalIgnoreCase)
|| string.Equals(header.Key, "X-Forwarded-For", StringComparison.OrdinalIgnoreCase))
continue;
var values = header.Value.ToArray();
@@ -188,9 +192,20 @@ public sealed class HttpStandbyRequestForwarder : IStandbyRequestForwarder
request.Content = content;
request.Headers.TryAddWithoutValidation(ForwardedRequestHeader, forwardedBy);
if (provenance is not null) request.Headers.TryAddWithoutValidation(RelayProvenance.Header, provenance);
return request;
}
private static bool KnownNotDispatched(HttpRequestException error)
=> error.HttpRequestError == HttpRequestError.NameResolutionError
|| error.InnerException is System.Net.Sockets.SocketException { SocketErrorCode: System.Net.Sockets.SocketError.ConnectionRefused };
private static IResult OutcomeUnknown() => Results.Json(new
{
error = "The peer may have applied this request. Verify its result before retrying.",
code = "OUTCOME_UNKNOWN"
}, statusCode: StatusCodes.Status502BadGateway);
private static IResult Unreachable() => Results.Json(
new
{
@@ -94,16 +94,20 @@ public sealed class InboundApiEndpointFilter : IEndpointFilter
if (!alreadyForwarded && forwarder is { IsConfigured: true })
{
httpContext.Items[RelayedToActiveItemKey] = true;
_logger.LogDebug("Inbound API request on a standby node — relaying to the active peer");
return await forwarder.ForwardAsync(httpContext, httpContext.RequestAborted)
.ConfigureAwait(false);
var result = await forwarder.ForwardAsync(httpContext, httpContext.RequestAborted).ConfigureAwait(false);
if (result is HttpStandbyRequestForwarder.RelayedResult)
httpContext.Items[RelayedToActiveItemKey] = true;
return result;
}
_logger.LogWarning(
alreadyForwarded
? "Inbound API request rejected — relayed here by a peer, but this node is also a standby (loop guard)"
: "Inbound API request rejected — this node is a standby (not the active central node) and no relay peers are configured");
// A verified internal refusal has not served the call. Its originating relay
// owns the single failure audit if all peers refuse; unsigned callers cannot suppress it.
if (RelayProvenance.IsVerified(httpContext)) httpContext.Items[RelayedToActiveItemKey] = true;
httpContext.Response.Headers[HttpStandbyRequestForwarder.StandbyResponseHeader] = "1";
return Results.Json(
new { error = "Inbound API is only available on the active central node", code = "STANDBY_NODE" },
@@ -63,10 +63,10 @@ public class InboundApiOptions
public List<string> StandbyForwardPeerEndpoints { get; set; } = new();
/// <summary>
/// Per-peer deadline for one relay attempt (connect + the active node's full method
/// execution). Must comfortably exceed <see cref="DefaultMethodTimeout"/> and the
/// longest per-method timeout in use, otherwise the standby gives up on a peer that
/// is still legitimately working and moves on to the next one. Default 60 s.
/// End-to-end relay budget spanning connection, permitted standby retries and response
/// consumption. Size it for the longest method execution plus network overhead. A send
/// whose response is lost or times out returns OUTCOME_UNKNOWN and is never replayed.
/// Default 60 s.
/// </summary>
public TimeSpan StandbyForwardTimeout { get; set; } = TimeSpan.FromSeconds(60);
@@ -107,6 +107,7 @@ public sealed class AuditWriteMiddleware
private readonly IOptionsMonitor<AuditLogOptions> _options;
private readonly IAuditActorAccessor? _actorAccessor;
private readonly IAuditInboundCeilingHitsCounter _ceilingHitsCounter;
private readonly IOptionsMonitor<InboundApiOptions>? _inboundOptions;
/// <summary>
/// Initializes the middleware with its required dependencies.
@@ -134,13 +135,15 @@ public sealed class AuditWriteMiddleware
ILogger<AuditWriteMiddleware> logger,
IOptionsMonitor<AuditLogOptions> options,
IAuditActorAccessor? actorAccessor = null,
IAuditInboundCeilingHitsCounter? ceilingHitsCounter = null)
IAuditInboundCeilingHitsCounter? ceilingHitsCounter = null,
IOptionsMonitor<InboundApiOptions>? inboundOptions = null)
{
_next = next ?? throw new ArgumentNullException(nameof(next));
_auditWriter = auditWriter ?? throw new ArgumentNullException(nameof(auditWriter));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_options = options ?? throw new ArgumentNullException(nameof(options));
_actorAccessor = actorAccessor;
_inboundOptions = inboundOptions;
_ceilingHitsCounter = ceilingHitsCounter ?? new NoOpAuditInboundCeilingHitsCounter();
}
@@ -151,6 +154,15 @@ public sealed class AuditWriteMiddleware
/// <returns>A task that represents the asynchronous operation.</returns>
public async Task InvokeAsync(HttpContext ctx)
{
// Install the route cap before EnableBuffering or any audit read makes Kestrel readonly.
var maxBytes = _inboundOptions?.CurrentValue.MaxRequestBodyBytes ?? new InboundApiOptions().MaxRequestBodyBytes;
var size = ctx.Features.Get<Microsoft.AspNetCore.Http.Features.IHttpMaxRequestBodySizeFeature>();
if (size is { IsReadOnly: false }) size.MaxRequestBodySize = maxBytes;
if (ctx.Request.ContentLength is { } declared && declared > maxBytes)
{
ctx.Response.StatusCode = StatusCodes.Status413PayloadTooLarge;
return;
}
var sw = Stopwatch.StartNew();
// Per-request hot read of the options snapshot so a live config change
@@ -169,6 +181,15 @@ public sealed class AuditWriteMiddleware
// ExecutionId (the row stays top-level — its ParentExecutionId is
// never set).
ctx.Items[InboundExecutionIdItemKey] = Guid.NewGuid();
try
{
if (_inboundOptions is not null) await RelayProvenance.VerifyAsync(ctx, _inboundOptions.CurrentValue);
}
catch (BadHttpRequestException ex) when (ex.StatusCode == StatusCodes.Status413PayloadTooLarge)
{
ctx.Response.StatusCode = StatusCodes.Status413PayloadTooLarge;
return;
}
// Buffer the request body up front so we can both audit it and let the
// downstream handler still parse it. EnableBuffering swaps the request
@@ -201,8 +222,15 @@ public sealed class AuditWriteMiddleware
if (!skipBody && RequestHasBody(ctx.Request))
{
ctx.Request.EnableBuffering();
(requestBody, requestTruncated) =
await ReadBufferedRequestBodyAsync(ctx.Request, cap).ConfigureAwait(false);
try
{
(requestBody, requestTruncated) = await ReadBufferedRequestBodyAsync(ctx.Request, cap).ConfigureAwait(false);
}
catch (BadHttpRequestException ex) when (ex.StatusCode == StatusCodes.Status413PayloadTooLarge)
{
ctx.Response.StatusCode = StatusCodes.Status413PayloadTooLarge;
return;
}
}
// Response body — wrap Response.Body in a forwarding stream that mirrors
@@ -322,6 +350,7 @@ public sealed class AuditWriteMiddleware
{
// Redact headers whose name appears in the HeaderRedactList —
// the same "<redacted>" marker used by ScadaBridgeAuditRedactor.
if (string.Equals(header.Key, RelayProvenance.Header, StringComparison.OrdinalIgnoreCase)) continue;
var value = redactSet.Contains(header.Key)
? "<redacted>"
: header.Value.ToString();
@@ -332,8 +361,9 @@ public sealed class AuditWriteMiddleware
// pipeline is the authoritative failure, so it wins; otherwise the endpoint may
// have stashed the method script's exception (which the executor keeps out of the
// HTTP body). Absent both, the row carries no detail.
var errorDetail = thrown?.ToString();
if (errorDetail is null
var suppressErrors = methodName is not null && opts.PerTargetOverrides.TryGetValue(methodName, out var privacy) && privacy.SkipBodyCapture;
var errorDetail = suppressErrors ? null : thrown?.ToString();
if (!suppressErrors && errorDetail is null
&& ctx.Items.TryGetValue(ErrorDetailItemKey, out var stashedErrorDetail))
{
errorDetail = stashedErrorDetail as string;
@@ -341,7 +371,7 @@ public sealed class AuditWriteMiddleware
var extra = JsonSerializer.Serialize(new
{
remoteIp = ctx.Connection.RemoteIpAddress?.ToString(),
remoteIp = RelayProvenance.CallerIp(ctx),
userAgent = ctx.Request.Headers.UserAgent.ToString(),
requestHeaders = headerDict,
});
@@ -367,7 +397,7 @@ public sealed class AuditWriteMiddleware
correlationId: null,
httpStatus: statusCode,
durationMs: (int)Math.Min(durationMs, int.MaxValue),
errorMessage: thrown?.Message,
errorMessage: suppressErrors ? null : thrown?.Message,
errorDetail: errorDetail,
requestSummary: requestBody,
responseSummary: responseBody,
@@ -512,6 +542,10 @@ public sealed class AuditWriteMiddleware
var content = DecodeUtf8Bounded(buffer, bytesForString, cutAtValidBytes: truncated);
return (string.IsNullOrEmpty(content) ? null : content, truncated);
}
catch (BadHttpRequestException ex) when (ex.StatusCode == StatusCodes.Status413PayloadTooLarge)
{
throw; // The route limit is authoritative, even when audit capture reads first.
}
catch
{
// A failed body read must not abort the request — fall through
@@ -0,0 +1,87 @@
using System.Net;
using System.Security.Cryptography;
using System.Text;
using System.Text.Json;
using Microsoft.AspNetCore.Http;
using ZB.MOM.WW.ScadaBridge.InboundAPI.Middleware;
namespace ZB.MOM.WW.ScadaBridge.InboundAPI;
/// <summary>One-hop caller provenance authenticated by the central pair's server-side pepper.</summary>
internal static class RelayProvenance
{
internal const string Header = "X-ScadaBridge-Relay";
private static readonly object VerifiedKey = new();
private sealed record Proof(long Time, Guid ExecutionId, string Ip, string Binding);
private static readonly byte[] Domain = "ScadaBridge.InboundRelay.v1"u8.ToArray();
internal static bool IsVerified(HttpContext context) => context.Items.ContainsKey(VerifiedKey);
internal static string? CallerIp(HttpContext context)
=> context.Items.TryGetValue(VerifiedKey, out var ip) ? ip as string : context.Connection.RemoteIpAddress?.ToString();
internal static string? Sign(HttpContext context, byte[] body, string pepper)
{
if (pepper.Length < 16 || context.Connection.RemoteIpAddress is null) return null;
var executionId = context.Items.TryGetValue(AuditWriteMiddleware.InboundExecutionIdItemKey, out var value) && value is Guid id
? id : Guid.NewGuid();
var proof = new Proof(DateTimeOffset.UtcNow.ToUnixTimeSeconds(), executionId,
context.Connection.RemoteIpAddress.ToString(), Binding(context.Request, body));
var bytes = JsonSerializer.SerializeToUtf8Bytes(proof);
return Convert.ToBase64String(bytes) + "." + Convert.ToHexString(Mac(bytes, pepper));
}
internal static async Task VerifyAsync(HttpContext context, InboundApiOptions options)
{
var header = context.Request.Headers[Header].ToString();
if (options.ApiKeyPepper.Length < 16 || header.Length is 0 or > 2048) return;
Proof? proof;
try
{
var parts = header.Split('.');
if (parts.Length != 2) return;
var payload = Convert.FromBase64String(parts[0]);
var signature = Convert.FromHexString(parts[1]);
if (!CryptographicOperations.FixedTimeEquals(signature, Mac(payload, options.ApiKeyPepper))) return;
proof = JsonSerializer.Deserialize<Proof>(payload);
if (proof is null || proof.ExecutionId == Guid.Empty || !IPAddress.TryParse(proof.Ip, out _)
|| Math.Abs((double)DateTimeOffset.UtcNow.ToUnixTimeSeconds() - proof.Time) > 300) return;
}
catch (Exception ex) when (ex is FormatException or JsonException or ArgumentException) { return; }
// Authenticate small metadata before reading; bound even non-Kestrel request streams.
context.Request.EnableBuffering();
var body = await ReadBoundedAsync(context.Request.Body, options.MaxRequestBodyBytes, context.RequestAborted);
context.Request.Body.Position = 0;
if (!string.Equals(proof.Binding, Binding(context.Request, body), StringComparison.Ordinal)) return;
context.Items[VerifiedKey] = proof.Ip;
context.Items[AuditWriteMiddleware.InboundExecutionIdItemKey] = proof.ExecutionId;
}
internal static async Task<byte[]> ReadBoundedAsync(Stream body, long limit, CancellationToken ct)
{
using var buffer = new MemoryStream();
var bytes = new byte[8192];
while (true)
{
var count = await body.ReadAsync(bytes.AsMemory(0, (int)Math.Min(bytes.Length, limit - buffer.Length + 1)), ct);
if (count == 0) return buffer.ToArray();
if (buffer.Length + count > limit)
throw new BadHttpRequestException("Request body too large", StatusCodes.Status413PayloadTooLarge);
await buffer.WriteAsync(bytes.AsMemory(0, count), ct);
}
}
private static string Binding(HttpRequest request, byte[] body)
{
var fields = new[] { request.Method, request.Path.Value + request.QueryString.Value,
request.ContentType ?? "", request.Headers.Authorization.ToString(), request.Headers["X-API-Key"].ToString(),
Convert.ToHexString(SHA256.HashData(body)) };
return Convert.ToHexString(SHA256.HashData(JsonSerializer.SerializeToUtf8Bytes(fields)));
}
private static byte[] Mac(byte[] payload, string pepper)
{
var key = HMACSHA256.HashData(Encoding.UTF8.GetBytes(pepper), Domain);
return HMACSHA256.HashData(key, payload);
}
}
@@ -240,14 +240,14 @@ public class RouteHelper
/// <summary>
/// Arch-review F11: returns a <see cref="RouteHelper"/> whose always-on <c>RouteCall</c> row
/// carries no request/response summary when <paramref name="captureBodies"/> is <c>false</c>.
/// carries no request/response summary or free-text error when <paramref name="captureBodies"/> is <c>false</c>.
/// <see cref="InboundScriptExecutor"/> sets this from the executing method's
/// <c>AuditLogOptions.PerTargetOverrides[method.Name].SkipBodyCapture</c> — the same override
/// <c>AuditWriteMiddleware</c> honours for the inbound row — so an operator who suppressed a
/// method's body from the audit log doesn't get it back via the routed call's arguments
/// (which are derived from that same body). Default <c>true</c> (unchanged behaviour).
/// </summary>
/// <param name="captureBodies">False to omit the request/response summary on the <c>RouteCall</c> row.</param>
/// <param name="captureBodies">False to omit request/response summaries and free-text errors on the <c>RouteCall</c> row.</param>
/// <returns>A new <see cref="RouteHelper"/> carrying the flag.</returns>
public RouteHelper WithBodyCapture(bool captureBodies) =>
new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId,
@@ -416,7 +416,7 @@ public class RouteTarget
executionId: _parentExecutionId,
parentExecutionId: null,
durationMs: (int)Math.Min(elapsedMs, int.MaxValue),
errorMessage: error,
errorMessage: _captureBodies ? error : null,
// Arch-review F11: a method with SkipBodyCapture set must not leak the routed
// call's arguments (derived from the same request body) or its return value.
requestSummary: _captureBodies ? TraceJson(args) : null,
@@ -300,7 +300,9 @@ public static class AuditEndpoints
}
var repo = context.RequestServices.GetRequiredService<IAuditLogRepository>();
var nodes = await repo.GetExecutionTreeAsync(executionId, context.RequestAborted);
var permitted = auth.User!.Roles.Contains("Administrator", StringComparer.OrdinalIgnoreCase)
|| auth.User.PermittedSiteIds.Length == 0 ? null : auth.User.PermittedSiteIds;
var nodes = await repo.GetExecutionTreeAsync(executionId, context.RequestAborted, permitted);
return Results.Json(nodes, JsonOptions);
}
@@ -536,21 +536,24 @@ public sealed class FlowDebugRunner
/// <summary>Replay (design §7.1): the stored ScriptTrace chain for any execution id.</summary>
public async Task<FlowDebugRunResult> ReplayAsync(
IServiceProvider sp, FlowDebugReplayCommand cmd, CancellationToken ct)
IServiceProvider sp, FlowDebugReplayCommand cmd, CancellationToken ct, AuthenticatedUser user)
{
ArgumentNullException.ThrowIfNull(sp);
ArgumentNullException.ThrowIfNull(cmd);
var startedAt = DateTimeOffset.UtcNow;
var events = await sp.GetRequiredService<IAuditLogRepository>()
.GetTraceEventsAsync(cmd.ExecutionId, ct).ConfigureAwait(false);
if (events.Count == 0)
.GetTraceEventsAsync(cmd.ExecutionId, ct, user.Roles.Contains("Administrator", StringComparer.OrdinalIgnoreCase) || user.PermittedSiteIds.Length == 0 ? null : user.PermittedSiteIds).ConfigureAwait(false);
if (events.Count == 0 && events is not ZB.MOM.WW.ScadaBridge.Commons.Types.Audit.TraceEventBatch { Truncated: true })
{
return new FlowDebugRunResult(cmd.ExecutionId, false,
$"No trace rows found for execution {cmd.ExecutionId:D}.", NotFoundCode,
events, false, startedAt, DateTimeOffset.UtcNow);
}
var rootId = AuditDetailsCodec.Deserialize(events[0].DetailsJson).ExecutionId ?? cmd.ExecutionId;
return new FlowDebugRunResult(rootId, true, null, null, events, false, startedAt, DateTimeOffset.UtcNow);
var batch = events as ZB.MOM.WW.ScadaBridge.Commons.Types.Audit.TraceEventBatch;
var rootId = batch?.RootExecutionId ?? (events.Count == 0 ? cmd.ExecutionId : AuditDetailsCodec.Deserialize(events[0].DetailsJson).ExecutionId ?? cmd.ExecutionId);
var collector = new FlowTraceCollector();
foreach (var evt in events) collector.Add(evt);
return new FlowDebugRunResult(rootId, true, null, null, collector.Complete(), batch?.Truncated == true || collector.Truncated, startedAt, DateTimeOffset.UtcNow);
}
/// <summary>
@@ -463,7 +463,7 @@ public class ManagementActor : ReceiveActor
FlowDebugRunMethodCommand cmd => await sp.GetRequiredService<FlowDebug.FlowDebugRunner>().RunMethodAsync(sp, cmd, user, CancellationToken.None),
FlowDebugRunValueChangeCommand cmd => await sp.GetRequiredService<FlowDebug.FlowDebugRunner>().RunValueChangeAsync(sp, cmd, user, CancellationToken.None),
FlowDebugRunScriptCommand cmd => await sp.GetRequiredService<FlowDebug.FlowDebugRunner>().RunScriptAsync(sp, cmd, user, CancellationToken.None),
FlowDebugReplayCommand cmd => await sp.GetRequiredService<FlowDebug.FlowDebugRunner>().ReplayAsync(sp, cmd, CancellationToken.None),
FlowDebugReplayCommand cmd => await sp.GetRequiredService<FlowDebug.FlowDebugRunner>().ReplayAsync(sp, cmd, CancellationToken.None, user),
// External Systems
ListExternalSystemsCommand => await HandleListExternalSystems(sp),
@@ -1020,10 +1020,17 @@ public class ManagementActor : ReceiveActor
{
EnforceSiteScope(user, cmd.SiteId);
var svc = sp.GetRequiredService<DeploymentService>();
var result = await svc.DeploySiteAsync(cmd.SiteId, user.Username, cmd.Force);
return result.IsSuccess
? result.Value
: throw new ManagementCommandException(result.Error);
var timeout = sp.GetService<Microsoft.Extensions.Options.IOptions<ManagementServiceOptions>>()?.Value.LongRunningCommandTimeout ?? TimeSpan.FromMinutes(5);
using var deadline = new CancellationTokenSource(timeout);
try
{
var result = await svc.DeploySiteAsync(cmd.SiteId, user.Username, cmd.Force, deadline.Token);
return result.IsSuccess ? result.Value : throw new ManagementCommandException(result.Error);
}
catch (OperationCanceledException) when (deadline.IsCancellationRequested)
{
throw new ManagementCommandException("Overall bulk deployment deadline exceeded before further apply could be scheduled.", "BULK_DEPLOY_TIMEOUT");
}
}
private static async Task<object?> HandleEnableInstance(IServiceProvider sp, MgmtEnableInstanceCommand cmd, AuthenticatedUser user)
@@ -74,6 +74,7 @@ public static class ManagementEndpoints
public static TimeSpan ResolveAskTimeout(ManagementServiceOptions? options, Type commandType)
=> LongRunningCommands.Contains(commandType)
? (options is { LongRunningCommandTimeout.Ticks: > 0 } o ? o.LongRunningCommandTimeout : DefaultLongRunningTimeout)
+ (commandType == typeof(MgmtDeploySiteCommand) ? TimeSpan.FromMinutes(1) : TimeSpan.Zero)
: ResolveAskTimeout(options);
/// <summary>Registers the <c>POST /management</c> endpoint on the given route builder.</summary>
@@ -837,7 +837,7 @@ public class AlarmActor : ReceiveActor
parentExecutionId,
// Scheduler seam (#18): share this alarm's scheduler override with the
// launched on-trigger script body (null = process-wide shared).
_scheduler);
_scheduler, Context.System.ActorSelection("/user/site-communication"));
}
catch (Exception ex)
{
@@ -129,9 +129,19 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
/// <summary>
/// Tracks Instance Actors that are terminating as part of a redeployment, keyed by
/// the terminating actor ref. The buffered command is applied once <see cref="Terminated"/>
/// confirms the child has fully stopped.
/// confirms the child has fully stopped and its final persistence acknowledgement arrives.
/// </summary>
private readonly Dictionary<IActorRef, PendingRedeploy> _pendingRedeploys = new();
private readonly Dictionary<IActorRef, InstanceActor.StaticOverridesFlushed> _redeployFlushes = new();
private readonly HashSet<IActorRef> _redeployTerminated = new();
private readonly HashSet<IActorRef> _redeployStopsInFlight = new();
private readonly Queue<IActorRef> _redeployStopQueue = new();
private sealed record LifecycleStop(string Name, string CommandId, bool Delete, IActorRef ReplyTo);
private sealed record LifecycleStopPersisted(IActorRef Actor, Exception? Error);
private readonly Dictionary<IActorRef, LifecycleStop> _lifecycleStops = new();
private readonly HashSet<string> _lifecycleStopNames = new();
/// <summary>
/// Name → terminating actor ref shadow of <see cref="_pendingRedeploys"/>.
@@ -487,6 +497,31 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
// Terminated signal — drains a buffered redeployment once the previous
// Instance Actor has fully stopped.
Receive<Terminated>(HandleTerminated);
Receive<LifecycleStopPersisted>(message =>
{
if (!_lifecycleStops.Remove(message.Actor, out var stop)) return;
_lifecycleStopNames.Remove(stop.Name);
_terminatingActorsByName.Remove(stop.Name);
BumpStoreWriteGeneration("lifecycle stop committed");
if (stop.Delete && message.Error is null)
{
_deployedInstanceNames.Remove(stop.Name);
_runningRevisions.Remove(stop.Name);
_sweepAttemptedRevisions.Remove(stop.Name);
}
UpdateInstanceCounts();
var action = stop.Delete ? "deleted" : "disabled";
LogDeploymentEvent(message.Error is null ? "Info" : "Error", stop.Name,
$"Instance {stop.Name} {action}", message.Error?.GetBaseException().Message);
stop.ReplyTo.Tell(new InstanceLifecycleResponse(stop.CommandId, stop.Name,
message.Error is null, message.Error?.GetBaseException().Message, DateTimeOffset.UtcNow));
});
Receive<InstanceActor.StaticOverridesFlushed>(message =>
{
if (!_pendingRedeploys.ContainsKey(message.Actor) && !_redeployStopsInFlight.Contains(message.Actor) && !_lifecycleStops.ContainsKey(message.Actor)) return;
_redeployFlushes[message.Actor] = message;
TryFinishRedeployStop(message.Actor);
});
}
/// <inheritdoc />
@@ -835,6 +870,13 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
private void HandleDeploy(DeployInstanceCommand command, IActorRef replyTo)
{
var instanceName = command.InstanceUniqueName;
if (_lifecycleStopNames.Contains(instanceName))
{
replyTo.Tell(new DeploymentStatusResponse(command.DeploymentId, instanceName, DeploymentStatus.Failed,
"Instance lifecycle persistence is in progress; retry after it completes.", DateTimeOffset.UtcNow));
return;
}
// WP3.1 warm-then-gate. The S3 gate below must stay synchronous on the actor thread
// (mailbox FIFO is what makes redeploy-supersede and delete-during-redeploy correct),
@@ -1029,7 +1071,8 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
_pendingRedeploys[existing] = new PendingRedeploy(command, replyTo);
_terminatingActorsByName[instanceName] = existing;
Context.Watch(existing);
Context.Stop(existing);
_redeployStopQueue.Enqueue(existing);
StartQueuedRedeployStops();
UpdateInstanceCounts();
return;
}
@@ -1061,6 +1104,74 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
ApplyDeployment(command, replyTo, isRedeploy: false);
}
private bool DeferLifecycleStop(string name, string commandId, bool delete, IActorRef replyTo)
{
if (_lifecycleStopNames.Contains(name))
{
replyTo.Tell(new InstanceLifecycleResponse(commandId, name, false,
"Instance lifecycle persistence is in progress; retry after it completes.", DateTimeOffset.UtcNow));
return true;
}
if (!_terminatingActorsByName.TryGetValue(name, out var actor) && !_instanceActors.TryGetValue(name, out actor))
return false;
if (_pendingRedeploys.Remove(actor, out var pending))
pending.OriginalSender.Tell(new DeploymentStatusResponse(pending.Command.DeploymentId, name,
DeploymentStatus.Failed, "Superseded by instance lifecycle command.", DateTimeOffset.UtcNow));
_instanceActors.Remove(name);
_terminatingActorsByName[name] = actor;
_lifecycleStopNames.Add(name);
_lifecycleStops[actor] = new LifecycleStop(name, commandId, delete, replyTo);
BumpStoreWriteGeneration("lifecycle stop dispatch");
Context.Watch(actor);
Context.Stop(actor);
UpdateInstanceCounts();
// Termination and flush may both already have arrived for a canceled
// redeploy. Reconcile immediately as well as on either completion signal.
TryFinishRedeployStop(actor);
return true;
}
private void StartQueuedRedeployStops()
{
while (_redeployStopsInFlight.Count < 4 && _redeployStopQueue.TryDequeue(out var actor))
{
if (!_pendingRedeploys.ContainsKey(actor)) continue;
_redeployStopsInFlight.Add(actor);
Context.Stop(actor);
}
}
private void TryFinishRedeployStop(IActorRef actor)
{
if (!_redeployTerminated.Contains(actor) || !_redeployFlushes.Remove(actor, out var flush)) return;
_redeployTerminated.Remove(actor);
_redeployStopsInFlight.Remove(actor);
if (_lifecycleStops.TryGetValue(actor, out var lifecycle))
{
var persistence = Task.Run(async () =>
{
if (flush.Error is not null) throw new InvalidOperationException("Final static override persistence failed.", flush.Error);
if (lifecycle.Delete) await _storage.RemoveDeployedConfigAsync(lifecycle.Name).ConfigureAwait(false);
else await _storage.SetInstanceEnabledAsync(lifecycle.Name, false).ConfigureAwait(false);
});
persistence.PipeTo(Self, Self, () => new LifecycleStopPersisted(actor, null),
error => new LifecycleStopPersisted(actor, error));
StartQueuedRedeployStops();
return;
}
if (_pendingRedeploys.Remove(actor, out var pending))
{
_terminatingActorsByName.Remove(pending.Command.InstanceUniqueName);
if (flush.Error is null)
ApplyDeployment(pending.Command, pending.OriginalSender, isRedeploy: true);
else
pending.OriginalSender.Tell(new DeploymentStatusResponse(pending.Command.DeploymentId,
pending.Command.InstanceUniqueName, DeploymentStatus.Failed,
"Predecessor static override flush failed; redeployment was not applied.", DateTimeOffset.UtcNow));
}
StartQueuedRedeployStops();
}
/// <summary>
/// Handles an Instance Actor's <see cref="Terminated"/> signal. Two cases:
/// <list type="number">
@@ -1075,16 +1186,10 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
{
// Case 1 — redeployment drain: the predecessor of a buffered redeploy has
// fully stopped and freed its actor name.
if (_pendingRedeploys.Remove(terminated.ActorRef, out var pending))
if (_pendingRedeploys.ContainsKey(terminated.ActorRef) || _redeployStopsInFlight.Contains(terminated.ActorRef) || _lifecycleStops.ContainsKey(terminated.ActorRef))
{
// Drop the name-keyed shadow now that the predecessor has
// fully terminated and its actor name is free. Any deploy arriving after
// this point sees neither _instanceActors[name] (cleared when we stopped
// the predecessor) nor _terminatingActorsByName[name] (cleared here), so
// ApplyDeployment + Context.ActorOf below safely reuses the name.
_terminatingActorsByName.Remove(pending.Command.InstanceUniqueName);
ApplyDeployment(pending.Command, pending.OriginalSender, isRedeploy: true);
_redeployTerminated.Add(terminated.ActorRef);
TryFinishRedeployStop(terminated.ActorRef);
return;
}
@@ -1437,33 +1542,7 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
// WP3.1 warm-then-gate: a deploy for this instance is mid-compile-warm; queue this
// command so same-instance ordering is preserved (see TryBufferDuringDeployWarm).
if (TryBufferDuringDeployWarm(instanceName, command, replyTo)) return;
// A disable arriving mid-redeploy must cancel the buffered
// redeploy. Otherwise HandleTerminated re-creates the Instance Actor and
// re-stores its config with isEnabled: true when the predecessor terminates,
// silently reverting the operator's disable back to enabled. Mirror the
// last-write-wins handling in HandleDeploy/HandleDelete: drop the pending
// command (so HandleTerminated returns early), clear the shadow, and tell the
// displaced deployer it was superseded. The disable itself still persists
// is_enabled = false below, which becomes the durable state.
if (_terminatingActorsByName.TryGetValue(instanceName, out var terminatingRef))
{
if (_pendingRedeploys.Remove(terminatingRef, out var pending))
{
pending.OriginalSender.Tell(new DeploymentStatusResponse(
pending.Command.DeploymentId,
instanceName,
DeploymentStatus.Failed,
$"superseded by disable of {instanceName} before redeploy finished terminating",
DateTimeOffset.UtcNow));
}
_terminatingActorsByName.Remove(instanceName);
}
else if (_instanceActors.TryGetValue(instanceName, out var actor))
{
Context.Stop(actor);
_instanceActors.Remove(instanceName);
}
if (DeferLifecycleStop(instanceName, command.CommandId, false, replyTo)) return;
UpdateInstanceCounts();
@@ -1512,6 +1591,13 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
private void HandleEnable(EnableInstanceCommand command, IActorRef replyTo)
{
var instanceName = command.InstanceUniqueName;
if (_lifecycleStopNames.Contains(instanceName))
{
replyTo.Tell(new InstanceLifecycleResponse(command.CommandId, instanceName, false,
"Instance lifecycle persistence is in progress; retry after it completes.", DateTimeOffset.UtcNow));
return;
}
// WP3.1 warm-then-gate: see TryBufferDuringDeployWarm.
if (TryBufferDuringDeployWarm(instanceName, command, replyTo)) return;
@@ -1997,37 +2083,7 @@ public class DeploymentManagerActor : ReceiveActor, IWithTimers
// WP3.1 warm-then-gate: see TryBufferDuringDeployWarm.
if (TryBufferDuringDeployWarm(instanceName, command, replyTo)) return;
// A delete arriving while a redeploy is still terminating must
// be authoritative over the mid-redeploy bookkeeping. HandleDeploy already
// removed the instance from _instanceActors and buffered a PendingRedeploy
// keyed by the terminating ref. If we fall straight through to the
// _instanceActors miss + unconditional decrement, the buffered redeploy is
// left intact — so when Terminated fires, HandleTerminated calls
// ApplyDeployment(isRedeploy: true) and RESURRECTS the just-deleted instance,
// with the counter now inconsistent. Cancel the pending redeploy first.
if (_terminatingActorsByName.TryGetValue(instanceName, out var terminatingRef))
{
// Drop the buffered command so HandleTerminated's _pendingRedeploys.Remove
// misses and it returns early (no resurrection). Clear the shadow too.
if (_pendingRedeploys.Remove(terminatingRef, out var pending))
{
pending.OriginalSender.Tell(new DeploymentStatusResponse(
pending.Command.DeploymentId,
instanceName,
DeploymentStatus.Failed,
$"superseded by delete of {instanceName} before redeploy finished terminating",
DateTimeOffset.UtcNow));
}
_terminatingActorsByName.Remove(instanceName);
// The terminating predecessor is already being stopped by HandleDeploy;
// no Context.Stop needed here.
}
else if (_instanceActors.TryGetValue(instanceName, out var actor))
{
Context.Stop(actor);
_instanceActors.Remove(instanceName);
}
if (DeferLifecycleStop(instanceName, command.CommandId, true, replyTo)) return;
// The deployed count is derived from the authoritative set of
// deployed config names, so removing the name here decrements it. Correct for a
@@ -120,15 +120,9 @@ public class InstanceActor : ReceiveActor
/// </summary>
private readonly Task? _staticStateResetGate;
/// <summary>
/// How long <see cref="PostStop"/> waits for the final coalesced static-override flush to
/// commit. The wait exists so a redeploy's clear is ordered AFTER this actor's last write
/// (arch-review 2026-08-19 P1-2 leg 2) — the Deployment Manager gates the clear on the
/// Terminated signal, which a fire-and-forget flush could outlive and durably resurrect
/// overrides the redeploy had just cleared. Bounded so a wedged SQLite write cannot stall
/// the redeploy indefinitely; exceeding it is logged and the stop proceeds.
/// </summary>
private static readonly TimeSpan FinalStaticOverrideFlushTimeout = TimeSpan.FromSeconds(5);
private Task _staticOverridePersistTail = Task.CompletedTask;
internal sealed record StaticOverridesFlushed(IActorRef Actor, Exception? Error);
/// <summary>
/// Cached string forms of <see cref="QualityCode"/> (perf remediation, arch-review
@@ -437,15 +431,13 @@ public class InstanceActor : ReceiveActor
state.Timer?.Cancel();
_tagSubscribeStates.Clear();
// Final flush of any coalesced static-override writes so a graceful stop does not lose
// the last <= flush-interval of writes (P4 pattern, mirrors NativeAlarmActor.PostStop).
// AWAITED, not fire-and-forget (arch-review 2026-08-19 P1-2 leg 2): a redeploy stops
// this actor, waits for Terminated, and only then clears the static overrides — so a
// detached final flush could commit up to one flush interval of the PREDECESSOR's dirty
// writes AFTER that clear and durably resurrect them. Blocking here is what puts the
// write provably before the Terminated signal the clear is gated on.
// SQLite's async APIs may block synchronously. Persist on the ordered
// thread-pool chain and acknowledge completion separately from Terminated.
_staticOverrideFlushTimer?.Cancel();
WaitForFinalStaticOverrideFlush();
var stoppedActor = Self;
FlushDirtyStaticOverrides(final: true).PipeTo(Context.Parent, Self,
success: () => new StaticOverridesFlushed(stoppedActor, null),
failure: error => new StaticOverridesFlushed(stoppedActor, error));
// Operational `instance_lifecycle` event — instance stopped. An
// instance stops on disable, delete, redeployment, and graceful shutdown;
@@ -659,15 +651,23 @@ public class InstanceActor : ReceiveActor
/// applied) remains authoritative.
/// </summary>
/// <returns>
/// The persist task, so the stop path can order the FINAL flush before termination
/// (<see cref="WaitForFinalStaticOverrideFlush"/>). Every other caller ignores it, keeping
/// The ordered persist task, acknowledged separately from termination so a redeploy
/// can wait for the final commit. Every other caller ignores it, keeping
/// the fire-and-forget semantics unchanged. Completed when there was nothing dirty.
/// </returns>
private Task FlushDirtyStaticOverrides()
private Task FlushDirtyStaticOverrides(bool final = false)
{
_staticOverrideFlushTimer = null;
if (_dirtyStaticOverrides.Count == 0)
return Task.CompletedTask;
return _staticOverridePersistTail;
if (!final && !_staticOverridePersistTail.IsCompleted)
{
// Keep coalescing while SQLite is busy instead of accumulating an
// unbounded task/batch queue behind a stalled transaction.
_staticOverrideFlushTimer = Context.System.Scheduler.ScheduleTellOnceCancelable(
_staticOverrideFlushInterval, Self, FlushStaticOverrides.Instance, Self);
return _staticOverridePersistTail;
}
var batch = _dirtyStaticOverrides
.Select(kv => (AttributeName: kv.Key, Value: kv.Value))
@@ -676,7 +676,14 @@ public class InstanceActor : ReceiveActor
var instanceName = _instanceUniqueName;
var logger = _logger;
var persist = _storage.SetStaticOverridesAsync(instanceName, batch);
var previous = _staticOverridePersistTail;
var persist = Task.Run(async () =>
{
try { await previous.ConfigureAwait(false); }
catch { /* The prior batch was logged; a transient failure must not poison future writes. */ }
await _storage.SetStaticOverridesAsync(instanceName, batch).ConfigureAwait(false);
});
_staticOverridePersistTail = persist;
persist.ContinueWith(t => logger.LogWarning(
t.Exception?.GetBaseException(),
"Failed to persist {Count} static overrides for {Instance}; in-memory state is authoritative",
@@ -685,34 +692,6 @@ public class InstanceActor : ReceiveActor
return persist;
}
/// <summary>
/// Runs the stop-path static-override flush and waits for it to commit, bounded by
/// <see cref="FinalStaticOverrideFlushTimeout"/>. A fault is swallowed — the
/// OnlyOnFaulted continuation inside <see cref="FlushDirtyStaticOverrides"/> has already
/// logged it, and a persist failure must never turn a graceful stop into a crash.
/// </summary>
private void WaitForFinalStaticOverrideFlush()
{
var flush = FlushDirtyStaticOverrides();
if (flush.IsCompleted)
return;
try
{
if (!flush.Wait(FinalStaticOverrideFlushTimeout))
{
_logger.LogWarning(
"Final static-override flush for {Instance} did not complete within {Timeout}; "
+ "stopping anyway — a redeploy's override reset may race it",
_instanceUniqueName, FinalStaticOverrideFlushTimeout);
}
}
catch (Exception)
{
// Already logged by the flush's OnlyOnFaulted continuation.
}
}
/// <summary>
/// Data-sourced attribute write: forwards a write request to the DCL and pipes
/// the device write result back to the caller. The in-memory value is left
@@ -86,6 +86,7 @@ public class NativeAlarmActor : ReceiveActor
/// a logging failure never affects the mirror.</param>
/// <param name="persistFlushInterval">How long native-alarm transitions coalesce before a
/// batched SQLite persist flush (P4). Defaults to 100 ms.</param>
/// <param name="rehydrationRead">Optional prestarted persisted-state read; null reads from storage.</param>
public NativeAlarmActor(
ResolvedNativeAlarmSource source,
string instanceName,
@@ -96,8 +97,10 @@ public class NativeAlarmActor : ReceiveActor
ILogger logger,
AlarmKind nativeKind = AlarmKind.NativeOpcUa,
IServiceProvider? serviceProvider = null,
TimeSpan? persistFlushInterval = null)
TimeSpan? persistFlushInterval = null,
Task<IReadOnlyList<NativeAlarmRow>>? rehydrationRead = null)
{
_rehydrationRead = rehydrationRead;
_source = source;
_instanceName = instanceName;
_instanceActor = instanceActor;
@@ -124,7 +127,7 @@ public class NativeAlarmActor : ReceiveActor
// Rehydrate last-known state from SQLite (non-blocking), then subscribe.
// A fresh source snapshot will reconcile/replace this once it arrives.
_storage.GetNativeAlarmsAsync(_instanceName, _source.CanonicalName)
(_rehydrationRead ?? _storage.GetNativeAlarmsAsync(_instanceName, _source.CanonicalName))
.PipeTo(Self, Self, rows => new RehydrationCompleted(rows));
SendSubscribe();
@@ -149,32 +152,48 @@ public class NativeAlarmActor : ReceiveActor
base.PostStop();
}
private readonly Task<IReadOnlyList<NativeAlarmRow>>? _rehydrationRead;
private string? _subscribeCorrelation;
private DateTimeOffset _subscribeDeadline;
private int _subscribeDelayMs;
private bool _subscribeSucceeded;
private void SendSubscribe()
{
_dclManager.Tell(new SubscribeAlarmsRequest(
CorrelationId: Guid.NewGuid().ToString("N"),
InstanceUniqueName: _instanceName,
ConnectionName: _source.ConnectionName,
SourceReference: _source.SourceReference,
ConditionFilter: _source.ConditionFilter,
Timestamp: DateTimeOffset.UtcNow));
// Arm a response-timeout retry so a LOST subscribe response is re-sent, not
// only a failure reply (S4). Cancelled on a Success reply; a failure reply
// re-arms it at 1× (overriding this 2× response-timeout arm). 2× interval so
// a slow-but-arriving response is not pre-empted by a redundant re-send.
if (_subscribeSucceeded) return;
if (_subscribeCorrelation is not null && DateTimeOffset.UtcNow < _subscribeDeadline)
{
_retryTimer?.Cancel();
_retryTimer = Context.System.Scheduler.ScheduleTellOnceCancelable(
TimeSpan.FromMilliseconds(Math.Max(1, (_subscribeDeadline - DateTimeOffset.UtcNow).TotalMilliseconds)),
Self, RetrySubscribe.Instance, Self);
return;
}
_subscribeDelayMs = _subscribeDelayMs == 0
? Math.Clamp(_options.NativeAlarmRetryIntervalMs, 1, 60000) : Math.Min(60000, _subscribeDelayMs * 2);
_subscribeCorrelation = Guid.NewGuid().ToString("N");
var responseTimeout = TimeSpan.FromMilliseconds(Math.Min(120000, _subscribeDelayMs * 2));
_subscribeDeadline = DateTimeOffset.UtcNow + responseTimeout;
_dclManager.Tell(new SubscribeAlarmsRequest(_subscribeCorrelation, _instanceName,
_source.ConnectionName, _source.SourceReference, _source.ConditionFilter, DateTimeOffset.UtcNow));
_retryTimer?.Cancel();
_retryTimer = Context.System.Scheduler.ScheduleTellOnceCancelable(
TimeSpan.FromMilliseconds(_options.NativeAlarmRetryIntervalMs * 2),
_retryTimer = Context.System.Scheduler.ScheduleTellOnceCancelable(responseTimeout,
Self, RetrySubscribe.Instance, Self);
}
private readonly HashSet<string> _liveTouched = new();
private bool _authoritativeSnapshotReceived;
private bool _rehydrationCompleted;
private void HandleRehydration(RehydrationCompleted msg)
{
if (_authoritativeSnapshotReceived || _rehydrationCompleted) return;
_rehydrationCompleted = true;
foreach (var row in msg.Rows)
{
// A live transition that arrived before the read completed wins.
if (_alarms.ContainsKey(row.SourceReference))
if (_alarms.ContainsKey(row.SourceReference) || _liveTouched.Contains(row.SourceReference))
{
continue;
}
@@ -227,8 +246,12 @@ public class NativeAlarmActor : ReceiveActor
// event (it is not a live transition).
Emit(t, t.Condition, logSiteEvent: false);
}
_liveTouched.Clear();
}
private bool _snapshotInProgress;
private readonly Dictionary<string, NativeAlarmTransition> _liveDuringSnapshot = new();
private void HandleTransition(NativeAlarmTransitionUpdate update)
{
var t = update.Transition;
@@ -237,15 +260,18 @@ public class NativeAlarmActor : ReceiveActor
switch (t.Kind)
{
case AlarmTransitionKind.SnapshotStart:
if (t.SnapshotId is null || t.SnapshotId == _lastCompletedSnapshotId || t.SnapshotId == _snapshotId)
if (t.SnapshotId is not null && (t.SnapshotId == _lastCompletedSnapshotId || t.SnapshotId == _snapshotId))
return;
_snapshotId = t.SnapshotId;
_snapshotInProgress = true;
_snapshotBuffer.Clear();
_liveDuringSnapshot.Clear();
break;
case AlarmTransitionKind.Snapshot:
if (t.SnapshotId != _snapshotId)
return;
_snapshotInProgress = true;
_snapshotBuffer[t.SourceReference] = t;
break;
@@ -270,6 +296,16 @@ public class NativeAlarmActor : ReceiveActor
/// </summary>
private void ApplySnapshotSwap()
{
foreach (var (source, live) in _liveDuringSnapshot)
{
if (_snapshotBuffer.TryGetValue(source, out var snapshot) && snapshot.TransitionTime > live.TransitionTime) continue;
if (!live.Condition.Active && live.Condition.Acknowledged) _snapshotBuffer.Remove(source);
else _snapshotBuffer[source] = live;
}
_liveDuringSnapshot.Clear();
_snapshotInProgress = false;
_authoritativeSnapshotReceived = true;
_liveTouched.Clear();
foreach (var (sourceRef, prior) in _alarms)
{
if (!_snapshotBuffer.ContainsKey(sourceRef))
@@ -304,12 +340,18 @@ public class NativeAlarmActor : ReceiveActor
private void ApplyLiveTransition(NativeAlarmTransition t)
{
if (!_authoritativeSnapshotReceived && !_rehydrationCompleted) _liveTouched.Add(t.SourceReference);
// Retain timestamp ordering for a clear removed from the visible mirror
// while a refresh is still assembling.
if (_liveDuringSnapshot.TryGetValue(t.SourceReference, out var recent) && t.TransitionTime < recent.TransitionTime)
return;
// Ignore stale (out-of-order) transitions for a condition we already hold.
if (_alarms.TryGetValue(t.SourceReference, out var existing) && t.TransitionTime < existing.TransitionTime)
{
return;
}
if (_snapshotInProgress) _liveDuringSnapshot[t.SourceReference] = t;
_alarms[t.SourceReference] = t;
MarkDirtyUpsert(t);
Emit(t, t.Condition);
@@ -333,6 +375,8 @@ public class NativeAlarmActor : ReceiveActor
// An interrupted refresh is not a snapshot. Never merge its partial rows
// into the next connection's replay.
_snapshotBuffer.Clear();
_liveDuringSnapshot.Clear();
_snapshotInProgress = false;
_snapshotId = null;
// Keep last-known conditions (uncertain) rather than clearing them; the
// reconnect snapshot reconciles state. No emit — avoids a flap.
@@ -343,8 +387,12 @@ public class NativeAlarmActor : ReceiveActor
private void HandleSubscribeResponse(SubscribeAlarmsResponse resp)
{
if (resp.CorrelationId != _subscribeCorrelation) return;
_subscribeCorrelation = null;
if (resp.Success)
{
_subscribeSucceeded = true;
_subscribeDelayMs = 0;
// Success — cancel the response-timeout retry armed in SendSubscribe.
_retryTimer?.Cancel();
_logger.LogInformation("Native alarm subscription established for {Source} on {Instance}",
@@ -358,7 +406,7 @@ public class NativeAlarmActor : ReceiveActor
_retryTimer?.Cancel();
_retryTimer = Context.System.Scheduler.ScheduleTellOnceCancelable(
TimeSpan.FromMilliseconds(_options.NativeAlarmRetryIntervalMs),
TimeSpan.FromMilliseconds(_subscribeDelayMs),
Self, RetrySubscribe.Instance, Self);
}
@@ -111,14 +111,13 @@ internal sealed class AuditingDbConnection : DbConnection
internal DbConnection Inner => _inner;
/// <inheritdoc />
// ConnectionString is settable on DbConnection — forward both halves.
// The configured target is the audit identity; scripts cannot retarget it or
// retrieve connection credentials through the raw ADO surface.
public override string ConnectionString
{
// Some providers throw on get when the connection hasn't been opened
// with a string set explicitly. The wrapper has no opinion — forward.
#pragma warning disable CS8765 // nullability of overridden member parameter — base setter accepts null in practice
get => _inner.ConnectionString;
set => _inner.ConnectionString = value;
#pragma warning disable CS8765
get => string.Empty;
set => throw new NotSupportedException("Audited connections cannot be retargeted.");
#pragma warning restore CS8765
}
@@ -135,7 +134,7 @@ internal sealed class AuditingDbConnection : DbConnection
public override ConnectionState State => _inner.State;
/// <inheritdoc />
public override void ChangeDatabase(string databaseName) => _inner.ChangeDatabase(databaseName);
public override void ChangeDatabase(string databaseName) => throw new NotSupportedException("Audited connections cannot be retargeted.");
/// <inheritdoc />
public override void Close() => _inner.Close();
@@ -165,6 +165,7 @@ internal static class ScriptRunLauncher
/// (external data has no spawning execution) — that on-trigger run is a tree ROOT.
/// </param>
/// <param name="scheduler">Script-execution scheduler seam (#18); null selects the process-wide shared instance.</param>
/// <param name="siteCommunicationActor">Site transport for notification status queries, resolved on the actor thread.</param>
public static void LaunchAlarmScript(
string alarmName,
string instanceName,
@@ -182,7 +183,8 @@ internal static class ScriptRunLauncher
long runId,
int? executionTimeoutSeconds,
Guid? parentExecutionId,
ScriptExecutionScheduler? scheduler)
ScriptExecutionScheduler? scheduler,
ICanTell? siteCommunicationActor = null)
{
var timeout = ResolveTimeout(executionTimeoutSeconds, options);
var executionScheduler = scheduler ?? ScriptExecutionScheduler.Shared(options);
@@ -196,7 +198,7 @@ internal static class ScriptRunLauncher
alarmName, instanceName, level, priority, message, compiledScript,
instanceActor, sharedScriptLibrary, options, completionTarget, logger,
healthCollector, serviceProvider, runId, parentExecutionId, timeout,
executionScheduler, cts),
executionScheduler, cts, siteCommunicationActor),
CancellationToken.None, TaskCreationOptions.DenyChildAttach, executionScheduler)
.Unwrap();
}
@@ -659,7 +661,8 @@ internal static class ScriptRunLauncher
Guid? parentExecutionId,
TimeSpan timeout,
ScriptExecutionScheduler executionScheduler,
CancellationTokenSource cts)
CancellationTokenSource cts,
ICanTell? siteCommunicationActor)
{
// arch-review 2026-08-19 P1-3: singletons resolved from the root provider up front,
// exactly like RunScriptAsync, so they are available to the catch blocks regardless
@@ -696,6 +699,9 @@ internal static class ScriptRunLauncher
// run it late — the alarm it would react to is already stale.
cts.Token.ThrowIfCancellationRequested();
using var serviceScope = serviceProvider?.CreateScope();
var services = serviceScope?.ServiceProvider;
// An alarm on-trigger run can call Instance.CallScript()
// via the ScriptRuntimeContext injected into globals
var context = new ScriptRuntimeContext(
@@ -711,6 +717,15 @@ internal static class ScriptRunLauncher
// it every Log.* line from an alarm script was attributed to the INSTANCE, not the
// alarm. `scriptName` above is already the "AlarmActor:{alarmName}" source key, so
// it is passed as-is (re-prefixing it would leave "AlarmActor:" in the target).
externalSystemClient: services?.GetService<IExternalSystemClient>(),
databaseGateway: services?.GetService<IDatabaseGateway>(),
storeAndForward: services?.GetService<StoreAndForwardService>(),
siteCommunicationActor: siteCommunicationActor,
siteId: services?.GetService<ISiteIdentityProvider>()?.SiteId ?? string.Empty,
auditWriter: services?.GetService<IAuditWriter>(),
operationTrackingStore: services?.GetService<IOperationTrackingStore>(),
cachedForwarder: services?.GetService<ICachedCallTelemetryForwarder>(),
sourceNode: services?.GetService<INodeIdentityProvider>()?.NodeName,
sourceScript: scriptName,
// ParentExecutionId tag-cascade: the
// alarm on-trigger run mints its own fresh ExecutionId (the
@@ -612,30 +612,11 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
/// Synchronously disposes the tracking store and its SQLite connection.
/// </summary>
/// <remarks>
/// This path does NOT bridge to async via
/// <c>.AsTask().GetAwaiter().GetResult()</c>. Sync-over-async on a SemaphoreSlim
/// can deadlock when invoked from a non-reentrant SyncContext (e.g. host
/// shutdown continuations observed on the host sync context). In-flight writes
/// at the moment of <see cref="Dispose"/> will fail their next operation
/// against the disposed connection with <see cref="ObjectDisposedException"/> —
/// the caller's responsibility is to ensure no concurrent operations during
/// the synchronous dispose. Use <see cref="DisposeAsync"/> if you need to
/// drain in-flight writes before close.
/// The pump runs on the thread pool and its drain does not capture a caller's
/// synchronization context. Join it before releasing SQLite resources, exactly
/// as the asynchronous path does; never close a live writer transaction.
/// </remarks>
public void Dispose()
{
if (Interlocked.Exchange(ref _disposeState, 1) != 0)
{
return;
}
// Stop accepting work; whatever is still queued fails on the pump with
// ObjectDisposedException (see ApplyBatch) — this path does not drain.
_queue.Writer.TryComplete();
_writeConnection.Dispose();
_writeGate.Dispose();
GC.SuppressFinalize(this);
}
public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();
/// <summary>
/// Asynchronously disposes the tracking store and its SQLite connection.
@@ -821,6 +821,16 @@ public class StoreAndForwardService
/// <see cref="_retryInProgress"/> flag (won the CAS); this method releases it in its
/// finally. Never call directly without holding ownership.
/// </summary>
private bool CanStartDelivery()
{
try { return _deliveryGate?.Invoke() ?? true; }
catch (Exception ex)
{
_logger.LogWarning(ex, "S&F delivery gate threw; no new attempt will start");
return false;
}
}
private async Task RunSweepOwnedAsync()
{
try
@@ -858,6 +868,7 @@ public class StoreAndForwardService
.Select(g => g.ToList())
.ToList();
var abandoned = 0;
using var laneCap = new SemaphoreSlim(Math.Max(1, _options.SweepTargetParallelism));
var laneTasks = lanes.Select(async lane =>
{
@@ -866,6 +877,12 @@ public class StoreAndForwardService
{
foreach (var message in lane)
{
if (Volatile.Read(ref abandoned) != 0) return;
if (!CanStartDelivery())
{
Interlocked.Exchange(ref abandoned, 1);
return;
}
// Short-circuit, per lane: one transient failure means
// the target is down — skip the rest of this lane this sweep
// rather than burning a full timeout per remaining message.
@@ -88,7 +88,8 @@ public class InstanceService
};
await _repository.AddInstanceAsync(instance, cancellationToken);
await _repository.SaveChangesAsync(cancellationToken);
try { await _repository.SaveChangesAsync(cancellationToken); }
catch (InstanceNameConflictException ex) { return Result<Instance>.Failure(ex.Message); }
await _auditService.LogAsync(user, "Create", "Instance", instance.Id.ToString(),
uniqueName, instance, cancellationToken);
@@ -4206,6 +4206,8 @@ public sealed class BundleImporter : IBundleImporter
case ResolutionAction.Rename:
{
var name = resolution.RenameTo ?? dto.UniqueName;
if (await _templateRepo.GetInstanceByUniqueNameAsync(name, ct).ConfigureAwait(false) is not null)
throw new InvalidOperationException($"An instance named '{name}' already exists. Choose a different rename target.");
var inst = BuildInstance(
dto, name, templateId, targetSite, areaId, connectionMaps);
await _templateRepo.AddInstanceAsync(inst, ct).ConfigureAwait(false);
@@ -4234,6 +4236,10 @@ public sealed class BundleImporter : IBundleImporter
break;
}
if (existing.SiteId != targetSite.Id && existing.State != InstanceState.NotDeployed)
throw new InvalidOperationException(
$"Instance '{existing.UniqueName}' must be undeployed before importing it to a different site.");
// Rewire scalar FKs + state, then replace child rows.
existing.TemplateId = templateId;
existing.SiteId = targetSite.Id;