ba7f8ee7a9
FlowDebugRunner.RunMethodAsync now passes InboundScriptResult.ErrorDetail through to the synthesized inbound audit row's errorDetail, so a SCRIPT_ERROR run's exception ToString() lands on AuditDetails.ErrorDetail for the Flow Debugger's inbound node to read (never surfaced to the HTTP caller). FlowDebugRunResult grows an additive LogsTruncated (default false, last positional member) reporting the inline FlowTraceCollector's low-priority (ScriptLog) truncation separately from Truncated (step-row fidelity); RunMethodAsync/RunScriptAsync thread collector.LogsTruncated through on their two collector-backed results.
637 lines
34 KiB
C#
637 lines
34 KiB
C#
using System.Diagnostics;
|
|
using System.Text.Json;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Logging;
|
|
using ZB.MOM.WW.Audit;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Scripts;
|
|
using ZB.MOM.WW.ScadaBridge.DeploymentManager;
|
|
using ZB.MOM.WW.ScadaBridge.InboundAPI;
|
|
using ZB.MOM.WW.ScadaBridge.Security;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.ManagementService.FlowDebug;
|
|
|
|
/// <summary>
|
|
/// Flow Debugger execution path (design 2026-09-05 §7.1): the single place a Live run or a
|
|
/// Replay happens, called by the per-node <c>ManagementActor</c>'s three
|
|
/// <c>FlowDebug*</c> handlers so the Central UI (via <c>ManagementActorHolder</c>) and the
|
|
/// CLI (via <c>/management</c>) share one path and one audit story. Resolves every
|
|
/// collaborator from the scoped <see cref="IServiceProvider"/> the actor hands it, so it is
|
|
/// unit-testable with substitutes and needs no actor.
|
|
/// </summary>
|
|
public sealed class FlowDebugRunner
|
|
{
|
|
/// <summary>Error code when a Live run is attempted on a standby central node.</summary>
|
|
public const string StandbyNodeCode = "STANDBY_NODE";
|
|
|
|
/// <summary>Error code when a value-change run spawns nothing inside its wait window.</summary>
|
|
public const string NoTriggerCode = "NO_TRIGGER_WITHIN_WINDOW";
|
|
|
|
/// <summary>Error code when the request parameters fail the method's schema.</summary>
|
|
public const string ValidationFailedCode = "VALIDATION_FAILED";
|
|
|
|
/// <summary>Error code when the method / instance / site does not exist.</summary>
|
|
public const string NotFoundCode = "NOT_FOUND";
|
|
|
|
/// <summary>Error code when a direct script call (design 2026-09-08 §4.4) fails at the site.</summary>
|
|
public const string CallFailedCode = "CALL_FAILED";
|
|
|
|
/// <summary>Error code when a Live run exceeds the method's / script's timeout plus <see cref="RunGrace"/>.</summary>
|
|
public const string TimeoutCode = "TIMEOUT";
|
|
|
|
/// <summary>
|
|
/// Timeout for a direct script call whose script declares none — mirrors
|
|
/// <c>SiteRuntimeOptions.ScriptExecutionTimeoutSeconds</c>'s default (the runner has no site options).
|
|
/// </summary>
|
|
public const int DefaultScriptTimeoutSeconds = 30;
|
|
|
|
/// <summary>
|
|
/// Ceiling on a write-kind run's spawned-run wait, whatever the command asks for. It must stay
|
|
/// UNDER <c>ManagementServiceOptions.LongRunningCommandTimeout</c> (5 minutes by default), which
|
|
/// is the Ask the management endpoint wraps this command in: a longer wait would leave the poll
|
|
/// loop running on the node after the caller's Ask has already expired, holding its scoped
|
|
/// provider and pulling from the site for a result nobody can receive.
|
|
/// </summary>
|
|
public const int MaxWaitSeconds = 300;
|
|
|
|
/// <summary>
|
|
/// Arch-review F2: <see cref="InboundScriptExecutor"/>'s code for a routed call refused
|
|
/// because the target site is outside the operator's site scope. InboundAPI cannot
|
|
/// reference this assembly, so it raises its own <c>RouteSiteScopeException</c> and the
|
|
/// runner converts the code back to <see cref="SiteScopeViolationException"/>.
|
|
/// </summary>
|
|
public const string SiteScopeDeniedCode = "SITE_SCOPE_DENIED";
|
|
|
|
/// <summary>The <c>Extra.source</c> value stamped on rows this runner synthesizes.</summary>
|
|
public const string SourceTag = "flow-debugger";
|
|
|
|
/// <summary>Grace added to a method's timeout before a Live run is reported as timed out (design §10).</summary>
|
|
public static readonly TimeSpan RunGrace = TimeSpan.FromSeconds(10);
|
|
|
|
/// <summary>How often the value-change run polls the site for spawned rows.</summary>
|
|
public static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(1);
|
|
|
|
/// <summary>Quiet period after the last new row before a value-change run is declared complete.</summary>
|
|
public static readonly TimeSpan SettleWindow = TimeSpan.FromSeconds(2);
|
|
|
|
private readonly ILogger<FlowDebugRunner> _logger;
|
|
private readonly Func<TimeSpan, CancellationToken, Task> _delay;
|
|
private readonly Func<DateTimeOffset> _now;
|
|
|
|
/// <summary>Creates the runner.</summary>
|
|
public FlowDebugRunner(ILogger<FlowDebugRunner> logger)
|
|
: this(logger, Task.Delay, () => DateTimeOffset.UtcNow)
|
|
{
|
|
}
|
|
|
|
/// <summary>Creates the runner with an injectable delay and clock (tests).</summary>
|
|
public FlowDebugRunner(
|
|
ILogger<FlowDebugRunner> logger,
|
|
Func<TimeSpan, CancellationToken, Task> delay,
|
|
Func<DateTimeOffset> now)
|
|
{
|
|
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
|
_delay = delay ?? throw new ArgumentNullException(nameof(delay));
|
|
_now = now ?? throw new ArgumentNullException(nameof(now));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Live run of an inbound API method (design §7.1): active-node gate, schema validation,
|
|
/// synthesized inbound row (actor = operator), traced execution through the real
|
|
/// <see cref="InboundScriptExecutor"/> with a trace-sink <see cref="RouteHelper"/>.
|
|
///
|
|
/// <para>
|
|
/// Arch-review F1: the sink is a <see cref="FlowTraceCollector"/> — the same collector the
|
|
/// site launcher fills — so the returned copy is REDACTED and capped by count and bytes,
|
|
/// and a Live result shows exactly the filtered rows a later Replay reads back. The
|
|
/// redactor is <c>AddAuditLog</c>'s <see cref="IAuditRedactor"/> singleton; a host that
|
|
/// registered none (a test composition root) collects raw but is still capped.
|
|
/// </para>
|
|
/// </summary>
|
|
public async Task<FlowDebugRunResult> RunMethodAsync(
|
|
IServiceProvider sp, FlowDebugRunMethodCommand cmd, AuthenticatedUser user, CancellationToken ct)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(sp);
|
|
ArgumentNullException.ThrowIfNull(cmd);
|
|
ArgumentNullException.ThrowIfNull(user);
|
|
EnsureActiveNode(sp);
|
|
|
|
var repo = sp.GetRequiredService<IInboundApiRepository>();
|
|
var method = await repo.GetApiMethodByIdAsync(cmd.ApiMethodId, ct).ConfigureAwait(false)
|
|
?? throw new ManagementCommandException($"ApiMethod with ID {cmd.ApiMethodId} not found.", NotFoundCode);
|
|
|
|
var resolveRef = await SchemaRefResolver.BuildAsync(
|
|
sp.GetService<ISharedSchemaRepository>(), [method.ParameterDefinitions], ct).ConfigureAwait(false);
|
|
var validation = ParameterValidator.Validate(cmd.Parameters, method.ParameterDefinitions, resolveRef);
|
|
if (!validation.IsValid)
|
|
{
|
|
throw new ManagementCommandException(
|
|
validation.ErrorMessage ?? "Parameter validation failed", ValidationFailedCode);
|
|
}
|
|
|
|
var root = Guid.NewGuid();
|
|
// One redactor for the whole run: the collector filters the emitted rows, and the runner
|
|
// filters the rows it synthesizes ITSELF (review follow-up) — the central writer redacts a
|
|
// private copy, so without this the caller's copy of the inbound row would be the raw one.
|
|
var redactor = sp.GetService<IAuditRedactor>();
|
|
var collector = new FlowTraceCollector(redactor: redactor);
|
|
var startedAt = DateTimeOffset.UtcNow;
|
|
var sw = Stopwatch.StartNew();
|
|
var timeout = TimeSpan.FromSeconds(Math.Max(1, method.TimeoutSeconds));
|
|
|
|
// Arch-review F2: a method Live run has no single target site to check up front the
|
|
// way RunValueChangeAsync does — the operator picks a METHOD, and where it routes is
|
|
// only known once the script runs. So the operator's scope is pushed into the
|
|
// RouteHelper and enforced at route time, before any site traffic.
|
|
var scope = await ResolvePermittedSiteIdentifiersAsync(sp, user, ct).ConfigureAwait(false);
|
|
var route = sp.GetRequiredService<RouteHelper>().WithTraceSink(evt => collector.Add(evt, lowPriority: FlowStepProjector.IsLogLine(evt))).WithSiteScope(scope);
|
|
// Step-logs D7: a ScriptLog row rides the collector's low-priority lane so the step rows
|
|
// emitted at run end are never crowded out of the inline copy (Replay stays complete).
|
|
var executor = sp.GetRequiredService<InboundScriptExecutor>();
|
|
|
|
InboundScriptResult result;
|
|
using (var runCts = CancellationTokenSource.CreateLinkedTokenSource(ct))
|
|
{
|
|
runCts.CancelAfter(timeout + RunGrace);
|
|
try
|
|
{
|
|
result = await executor.ExecuteAsync(
|
|
method, validation.Parameters, route, timeout, runCts.Token,
|
|
parentExecutionId: root, trace: true).ConfigureAwait(false);
|
|
}
|
|
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
|
|
{
|
|
result = new InboundScriptResult(false, null, TimeoutMessage(timeout), TimeoutCode);
|
|
}
|
|
}
|
|
|
|
var finishedAt = DateTimeOffset.UtcNow;
|
|
// Close the sink here, not at the point the list is built: an un-awaited call finishing
|
|
// after the run returned must not mutate the snapshot the reply is assembled from.
|
|
var events = collector.Complete();
|
|
var requestJson = cmd.Parameters is { } p ? p.GetRawText() : null;
|
|
var inboundRow = ScadaBridgeAuditEventFactory.Create(
|
|
AuditChannel.ApiInbound, AuditKind.InboundRequest,
|
|
result.Success ? AuditStatus.Delivered : AuditStatus.Failed,
|
|
occurredAtUtc: startedAt.UtcDateTime,
|
|
actor: user.Username,
|
|
target: method.Name,
|
|
sourceNode: sp.GetService<INodeIdentityProvider>()?.NodeName,
|
|
executionId: root,
|
|
durationMs: (int)Math.Min(sw.ElapsedMilliseconds, int.MaxValue),
|
|
errorMessage: result.ErrorMessage,
|
|
errorDetail: result.ErrorDetail,
|
|
requestSummary: requestJson,
|
|
responseSummary: result.ResultJson,
|
|
extra: new ScriptTraceExtra { Source = SourceTag }.ToJson());
|
|
await WriteCentralAsync(sp, inboundRow, ct).ConfigureAwait(false);
|
|
|
|
// Arch-review F2: convert the executor's code into the same authorization failure
|
|
// RunValueChangeAsync raises, so both Live entry points surface one exception type
|
|
// (→ ManagementUnauthorized). Deliberately AFTER the inbound row is written: a
|
|
// denied attempt must still be audited against the operator's name.
|
|
if (string.Equals(result.ErrorCode, SiteScopeDeniedCode, StringComparison.Ordinal))
|
|
{
|
|
throw new SiteScopeViolationException(
|
|
result.ErrorMessage ?? "Access denied: routed call outside your site scope.");
|
|
}
|
|
|
|
// Arch-review F5: the collector's emission order is authoritative — central rows carry central's
|
|
// clock and site rows the site's, so re-sorting by OccurredAtUtc lets a skewed site clock float
|
|
// its steps ahead of the RouteCall that started them. (The value-change path below keeps its
|
|
// OrderBy: every row it merges was stamped by the same site.)
|
|
var ordered = new List<AuditEvent>(events.Count + 1) { redactor?.Apply(inboundRow) ?? inboundRow };
|
|
ordered.AddRange(events);
|
|
|
|
return new FlowDebugRunResult(
|
|
root, result.Success, result.ErrorMessage, result.Success ? null : result.ErrorCode,
|
|
ordered, Truncated: collector.Truncated, startedAt, finishedAt, LogsTruncated: collector.LogsTruncated);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Live run of a value-change trigger (design §5.3/§5.4): traced routed write through the
|
|
/// site's normal path, then poll the site's trace rows for the spawned run(s).
|
|
/// </summary>
|
|
public async Task<FlowDebugRunResult> RunValueChangeAsync(
|
|
IServiceProvider sp, FlowDebugRunValueChangeCommand cmd, AuthenticatedUser user, CancellationToken ct)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(sp);
|
|
ArgumentNullException.ThrowIfNull(cmd);
|
|
ArgumentNullException.ThrowIfNull(user);
|
|
if (string.IsNullOrWhiteSpace(cmd.AttributeName))
|
|
throw new ManagementCommandException("AttributeName is required.", ValidationFailedCode);
|
|
EnsureActiveNode(sp);
|
|
|
|
var instance = await sp.GetRequiredService<ITemplateEngineRepository>()
|
|
.GetInstanceByIdAsync(cmd.InstanceId, ct).ConfigureAwait(false)
|
|
?? throw new ManagementCommandException($"Instance with ID {cmd.InstanceId} not found.", NotFoundCode);
|
|
EnforceSiteScope(user, instance.SiteId);
|
|
var site = await sp.GetRequiredService<ISiteRepository>()
|
|
.GetSiteByIdAsync(instance.SiteId, ct).ConfigureAwait(false)
|
|
?? throw new ManagementCommandException($"Site with ID {instance.SiteId} not found.", NotFoundCode);
|
|
|
|
return await WriteAndPollAsync(
|
|
sp, user, instance, site, cmd.AttributeName, cmd.Value, cmd.WaitSeconds, triggerScript: null, ct)
|
|
.ConfigureAwait(false);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Live run of a site script by name (design 2026-09-08 §4): resolves the script's effective
|
|
/// definition from the flattened instance configuration and branches on its trigger kind —
|
|
/// write kinds go through <see cref="WriteAndPollAsync"/>, every other kind is a traced
|
|
/// <see cref="IInstanceRouter.RouteToCallAsync"/> with one operator-attributed
|
|
/// <see cref="AuditKind.RouteCall"/> root row. Calling <see cref="IInstanceRouter"/> directly
|
|
/// rather than through <see cref="RouteHelper"/> is deliberate: the helper emits its own
|
|
/// always-on RouteCall row with no operator actor, which would give the graph two roots.
|
|
/// </summary>
|
|
public async Task<FlowDebugRunResult> RunScriptAsync(
|
|
IServiceProvider sp, FlowDebugRunScriptCommand cmd, AuthenticatedUser user, CancellationToken ct)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(sp);
|
|
ArgumentNullException.ThrowIfNull(cmd);
|
|
ArgumentNullException.ThrowIfNull(user);
|
|
if (string.IsNullOrWhiteSpace(cmd.ScriptName))
|
|
throw new ManagementCommandException("ScriptName is required.", ValidationFailedCode);
|
|
EnsureActiveNode(sp);
|
|
|
|
var instance = await sp.GetRequiredService<ITemplateEngineRepository>()
|
|
.GetInstanceByIdAsync(cmd.InstanceId, ct).ConfigureAwait(false)
|
|
?? throw new ManagementCommandException($"Instance with ID {cmd.InstanceId} not found.", NotFoundCode);
|
|
EnforceSiteScope(user, instance.SiteId);
|
|
var site = await sp.GetRequiredService<ISiteRepository>()
|
|
.GetSiteByIdAsync(instance.SiteId, ct).ConfigureAwait(false)
|
|
?? throw new ManagementCommandException($"Site with ID {instance.SiteId} not found.", NotFoundCode);
|
|
|
|
// Design §2: the flattened configuration is the one place canonical names, effective triggers and
|
|
// effective parameter definitions all live (composition + inheritance + instance overrides applied).
|
|
var flattened = await sp.GetRequiredService<IFlatteningPipeline>()
|
|
.FlattenAndValidateAsync(instance.Id, ct, validateScripts: false).ConfigureAwait(false);
|
|
if (flattened.IsFailure)
|
|
throw new ManagementCommandException($"Could not resolve instance '{instance.UniqueName}': {flattened.Error}", NotFoundCode);
|
|
var script = flattened.Value.Configuration.Scripts
|
|
.FirstOrDefault(s => string.Equals(s.CanonicalName, cmd.ScriptName, StringComparison.OrdinalIgnoreCase))
|
|
?? throw new ManagementCommandException($"Script '{cmd.ScriptName}' is not defined on instance '{instance.UniqueName}'.", NotFoundCode);
|
|
|
|
if (ScriptTriggerAttribute.IsWriteKind(script.TriggerType))
|
|
{
|
|
if (!ScriptTriggerAttribute.TryRead(script.TriggerType, script.TriggerConfiguration, out var attribute))
|
|
{
|
|
throw new ManagementCommandException(
|
|
$"Script '{script.CanonicalName}' has a {script.TriggerType} trigger with no readable attribute; fix its trigger configuration.",
|
|
ValidationFailedCode);
|
|
}
|
|
|
|
if (cmd.Value is null)
|
|
throw new ManagementCommandException($"Value is required to fire '{script.CanonicalName}' (writes '{attribute}').", ValidationFailedCode);
|
|
return await WriteAndPollAsync(
|
|
sp, user, instance, site, attribute!, cmd.Value, cmd.WaitSeconds, script.CanonicalName, ct).ConfigureAwait(false);
|
|
}
|
|
|
|
var resolveRef = await SchemaRefResolver.BuildAsync(
|
|
sp.GetService<ISharedSchemaRepository>(), [script.ParameterDefinitions], ct).ConfigureAwait(false);
|
|
var validation = ParameterValidator.Validate(cmd.Parameters, script.ParameterDefinitions, resolveRef);
|
|
if (!validation.IsValid)
|
|
throw new ManagementCommandException(validation.ErrorMessage ?? "Parameter validation failed", ValidationFailedCode);
|
|
|
|
var root = Guid.NewGuid();
|
|
var startedAt = DateTimeOffset.UtcNow;
|
|
var sw = Stopwatch.StartNew();
|
|
// Review follow-up: the root row is synthesized here, so the copy handed back must be
|
|
// redacted by this method — WriteCentralAsync's writer only redacts its own private copy.
|
|
var redactor = sp.GetService<IAuditRedactor>();
|
|
var collector = new FlowTraceCollector(redactor: redactor);
|
|
var timeout = TimeSpan.FromSeconds(Math.Max(1, script.ExecutionTimeoutSeconds ?? DefaultScriptTimeoutSeconds));
|
|
var router = sp.GetRequiredService<IInstanceRouter>();
|
|
var request = new RouteToCallRequest(
|
|
Guid.NewGuid().ToString(), instance.UniqueName, script.CanonicalName,
|
|
validation.Parameters.Count == 0 ? null : validation.Parameters,
|
|
DateTimeOffset.UtcNow, ParentExecutionId: root, Trace: true);
|
|
|
|
RouteToCallResponse response;
|
|
// Null while the site ANSWERED (a success or a Success=false reply); otherwise the code to
|
|
// report for a routed Ask that threw instead of answering.
|
|
string? faultCode = null;
|
|
using (var runCts = CancellationTokenSource.CreateLinkedTokenSource(ct))
|
|
{
|
|
runCts.CancelAfter(timeout + RunGrace);
|
|
try
|
|
{
|
|
response = await router.RouteToCallAsync(site.SiteIdentifier, request, runCts.Token).ConfigureAwait(false);
|
|
}
|
|
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
|
|
{
|
|
faultCode = TimeoutCode;
|
|
response = new RouteToCallResponse(
|
|
request.CorrelationId, false, null, TimeoutMessage(timeout), DateTimeOffset.UtcNow);
|
|
}
|
|
catch (Exception ex) when (ex is not OperationCanceledException)
|
|
{
|
|
// RouteToCallAsync is an Akka Ask bounded by CommunicationOptions.IntegrationTimeout,
|
|
// so it throws AskTimeoutException (NOT an OperationCanceledException) when the site
|
|
// never answers — and any transport fault throws too. "No answer" is NOT "did not
|
|
// happen": the site may already have run the script. Letting the exception escape
|
|
// would lose the root row entirely and show the operator MapFault's generic internal
|
|
// error for a call that ran, so it is converted into a failed response the same way
|
|
// RouteHelper.CallAsync types its Ask faults and WriteAndPollAsync reports a failed
|
|
// write. A cancellation of the CALLER's token still propagates (filtered out above).
|
|
faultCode = ex is Akka.Actor.AskTimeoutException ? TimeoutCode : CallFailedCode;
|
|
_logger.LogWarning(ex,
|
|
"Flow Debugger routed call to site {Site} for {Instance}.{Script} faulted",
|
|
site.SiteIdentifier, instance.UniqueName, script.CanonicalName);
|
|
response = new RouteToCallResponse(
|
|
request.CorrelationId, false, null, ex.Message, DateTimeOffset.UtcNow);
|
|
}
|
|
}
|
|
|
|
// Everything from here to WriteCentralAsync runs AFTER the site already executed the
|
|
// script — a real side effect. Nothing in this stretch may throw, or the root row is never
|
|
// written and the operator sees ManagementActor.MapFault's generic "internal error" for a
|
|
// call that actually succeeded. So the trace drain skips a faulting event and the return
|
|
// value falls back to a placeholder, the same way InboundScriptExecutor turns a script's
|
|
// own faults into a captured error rather than an escaping exception.
|
|
foreach (var e in response.TraceEvents ?? [])
|
|
{
|
|
try
|
|
{
|
|
collector.Add(e, lowPriority: FlowStepProjector.IsLogLine(e));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex,
|
|
"Flow Debugger dropped an inline trace event from the run of {Script} on {Instance}",
|
|
script.CanonicalName, instance.UniqueName);
|
|
}
|
|
}
|
|
|
|
var events = collector.Complete();
|
|
|
|
var rootRow = ScadaBridgeAuditEventFactory.Create(
|
|
AuditChannel.ScriptTrace, AuditKind.RouteCall,
|
|
response.Success ? AuditStatus.Delivered : AuditStatus.Failed,
|
|
occurredAtUtc: startedAt.UtcDateTime,
|
|
actor: user.Username,
|
|
target: $"{instance.UniqueName}.{script.CanonicalName}",
|
|
sourceNode: sp.GetService<INodeIdentityProvider>()?.NodeName,
|
|
executionId: root,
|
|
sourceSiteId: site.SiteIdentifier,
|
|
sourceInstanceId: instance.UniqueName,
|
|
durationMs: (int)Math.Min(sw.ElapsedMilliseconds, int.MaxValue),
|
|
errorMessage: response.ErrorMessage,
|
|
requestSummary: cmd.Parameters is { } p ? p.GetRawText() : null,
|
|
responseSummary: SerializeReturnValue(response.ReturnValue),
|
|
extra: new ScriptTraceExtra { Source = SourceTag, ChildExecutionId = response.ChildExecutionId }.ToJson());
|
|
await WriteCentralAsync(sp, rootRow, ct).ConfigureAwait(false);
|
|
|
|
// Arch-review F5: emission order is authoritative — the root row carries central's clock and
|
|
// the site rows the site's, so re-sorting by OccurredAtUtc lets a skewed site clock float its
|
|
// steps ahead of the RouteCall that started them.
|
|
var ordered = new List<AuditEvent>(events.Count + 1) { redactor?.Apply(rootRow) ?? rootRow };
|
|
ordered.AddRange(events);
|
|
return new FlowDebugRunResult(
|
|
root, response.Success, response.ErrorMessage,
|
|
response.Success ? null : faultCode ?? CallFailedCode,
|
|
ordered, Truncated: collector.Truncated, startedAt, DateTimeOffset.UtcNow,
|
|
LogsTruncated: collector.LogsTruncated);
|
|
}
|
|
|
|
/// <summary>
|
|
/// The traced attribute write + spawned-run poll shared by <see cref="RunValueChangeAsync"/> and
|
|
/// the write-kind branch of the script entry's <c>RunScriptAsync</c> (design 2026-09-08 §4.3). The
|
|
/// caller has already gated the active node, loaded and scope-checked the instance, and loaded the
|
|
/// site. <paramref name="triggerScript"/> is null for the attribute-level command and the script's
|
|
/// canonical name for the script entry — it only changes the NO_TRIGGER message.
|
|
/// </summary>
|
|
private async Task<FlowDebugRunResult> WriteAndPollAsync(
|
|
IServiceProvider sp, AuthenticatedUser user, Instance instance, Site site,
|
|
string attributeName, string? value, int waitSeconds, string? triggerScript, CancellationToken ct)
|
|
{
|
|
var root = Guid.NewGuid();
|
|
var startedAt = DateTimeOffset.UtcNow;
|
|
var sw = Stopwatch.StartNew();
|
|
// Review follow-up: the root row is synthesized here, so the copy handed back must be
|
|
// redacted by this method — WriteCentralAsync's writer only redacts its own private copy.
|
|
var redactor = sp.GetService<IAuditRedactor>();
|
|
// IInstanceRouter is the same seam RouteHelper uses (central binds it to the comms layer).
|
|
var router = sp.GetRequiredService<IInstanceRouter>();
|
|
var write = await router.RouteToSetAttributesAsync(
|
|
site.SiteIdentifier,
|
|
new RouteToSetAttributesRequest(
|
|
Guid.NewGuid().ToString(), instance.UniqueName,
|
|
new Dictionary<string, string> { [attributeName] = value ?? string.Empty },
|
|
DateTimeOffset.UtcNow, ParentExecutionId: root, Trace: true),
|
|
ct).ConfigureAwait(false);
|
|
|
|
var rootRow = ScadaBridgeAuditEventFactory.Create(
|
|
AuditChannel.ScriptTrace, AuditKind.AttributeWrite,
|
|
write.Success ? AuditStatus.Delivered : AuditStatus.Failed,
|
|
occurredAtUtc: startedAt.UtcDateTime,
|
|
actor: user.Username,
|
|
target: attributeName,
|
|
sourceNode: sp.GetService<INodeIdentityProvider>()?.NodeName,
|
|
executionId: root,
|
|
sourceSiteId: site.SiteIdentifier,
|
|
sourceInstanceId: instance.UniqueName,
|
|
durationMs: (int)Math.Min(sw.ElapsedMilliseconds, int.MaxValue),
|
|
errorMessage: write.ErrorMessage,
|
|
requestSummary: JsonSerializer.Serialize(new { attribute = attributeName, value = value }, ScriptTraceExtra.JsonOptions),
|
|
extra: new ScriptTraceExtra
|
|
{
|
|
Source = SourceTag,
|
|
Writes = [new AttributeWriteDetail(attributeName, value, write.Success, write.ErrorMessage)],
|
|
}.ToJson());
|
|
await WriteCentralAsync(sp, rootRow, ct).ConfigureAwait(false);
|
|
var redactedRoot = redactor?.Apply(rootRow) ?? rootRow;
|
|
|
|
if (!write.Success)
|
|
{
|
|
return new FlowDebugRunResult(root, false, write.ErrorMessage ?? "Attribute write failed", "WRITE_FAILED",
|
|
[redactedRoot], false, startedAt, DateTimeOffset.UtcNow);
|
|
}
|
|
|
|
// The run that COUNTS as completion. For the script entry (design 2026-09-08 §4.3) that is
|
|
// the script the operator chose, matched on Target — the site stamps ScriptRunLauncher's
|
|
// `target: scriptName` on the row. Other scripts the same write happens to fire are still
|
|
// collected below (the graph draws the siblings) but must never report the CHOSEN script as
|
|
// having run. triggerScript == null is the attribute-level command, which expects no
|
|
// particular script, so any spawned run satisfies it.
|
|
bool IsAwaitedRun(AuditEvent e)
|
|
{
|
|
var d = AuditDetailsCodec.Deserialize(e.DetailsJson);
|
|
if (d.Kind != nameof(AuditKind.ScriptRun) || d.ParentExecutionId != root) return false;
|
|
return triggerScript is null
|
|
|| string.Equals(e.Target, triggerScript, StringComparison.OrdinalIgnoreCase);
|
|
}
|
|
|
|
// Poll the site for the traced run(s) the write spawned — matched by ParentExecutionId
|
|
// == root, then their children — until the awaited ScriptRun exists AND the set has been
|
|
// quiet for SettleWindow, or the (clamped) wait elapses.
|
|
var puller = sp.GetRequiredService<IPullTraceEventsClient>();
|
|
var wait = Math.Clamp(waitSeconds, 1, MaxWaitSeconds);
|
|
var deadline = _now() + TimeSpan.FromSeconds(wait);
|
|
var known = new Dictionary<Guid, AuditEvent>();
|
|
DateTimeOffset? lastNew = null;
|
|
var since = startedAt.UtcDateTime.AddSeconds(-5);
|
|
while (!ct.IsCancellationRequested)
|
|
{
|
|
var ids = new HashSet<Guid> { root };
|
|
foreach (var e in known.Values)
|
|
{
|
|
var d = AuditDetailsCodec.Deserialize(e.DetailsJson);
|
|
if (d.ExecutionId is { } x) ids.Add(x);
|
|
}
|
|
var pulled = await puller.PullAsync(site.SiteIdentifier, ids.Take(64).ToList(), since, 500, ct).ConfigureAwait(false);
|
|
var added = false;
|
|
foreach (var e in pulled)
|
|
{
|
|
if (known.TryAdd(e.EventId, e)) added = true;
|
|
}
|
|
if (added) lastNew = _now();
|
|
|
|
var spawned = known.Values.Any(IsAwaitedRun);
|
|
var now = _now();
|
|
if (spawned && lastNew is { } ln && now - ln >= SettleWindow) break;
|
|
if (now >= deadline) break;
|
|
await _delay(PollInterval, ct).ConfigureAwait(false);
|
|
}
|
|
|
|
var ordered = new List<AuditEvent> { redactedRoot };
|
|
ordered.AddRange(known.Values.OrderBy(e => e.OccurredAtUtc).ThenBy(e => e.EventId));
|
|
var completed = known.Values.Any(IsAwaitedRun);
|
|
return new FlowDebugRunResult(
|
|
root, completed,
|
|
completed ? null : triggerScript is null
|
|
? $"No traced run was spawned within {wait}s (no value-change trigger on '{attributeName}', or no device echo)."
|
|
: $"No traced run of '{triggerScript}' was spawned within {wait}s after writing '{attributeName}' (trigger condition not met, or no device echo).",
|
|
completed ? null : NoTriggerCode,
|
|
ordered, false, startedAt, DateTimeOffset.UtcNow);
|
|
}
|
|
|
|
/// <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)
|
|
{
|
|
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)
|
|
{
|
|
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);
|
|
}
|
|
|
|
/// <summary>
|
|
/// The timed-out message shared by both Live entry points, so the two paths never drift.
|
|
/// </summary>
|
|
private static string TimeoutMessage(TimeSpan timeout) =>
|
|
$"Flow Debugger run timed out after {timeout.TotalSeconds}s + {RunGrace.TotalSeconds}s grace";
|
|
|
|
/// <summary>
|
|
/// Serializes a routed call's return value for the root row's <c>ResponseSummary</c>, NEVER
|
|
/// throwing: <see cref="ScriptTraceExtra.JsonOptions"/> rejects named floating-point literals,
|
|
/// so a script returning <c>double.NaN</c> / an infinity — or any other value the serializer
|
|
/// cannot represent — would otherwise take down the audit write of a call that already ran.
|
|
/// The placeholder keeps the row (and the operator's Live result) intact.
|
|
///
|
|
/// <para>
|
|
/// The catch is deliberately broad (as <see cref="WriteCentralAsync"/>'s already is): the value
|
|
/// crossing this seam came from a user-authored script, and the serializer's failure mode varies
|
|
/// by value — <c>ArgumentException</c> for a non-finite number, <see cref="JsonException"/> for a
|
|
/// cycle, <see cref="NotSupportedException"/> for an unmappable type. A best-effort audit summary
|
|
/// must not be the thing that fails a call the site already ran.
|
|
/// </para>
|
|
/// </summary>
|
|
private static string? SerializeReturnValue(object? returnValue)
|
|
{
|
|
if (returnValue is null) return null;
|
|
try
|
|
{
|
|
return JsonSerializer.Serialize(returnValue, ScriptTraceExtra.JsonOptions);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
return $"<unserializable return value: {ex.GetType().Name}>";
|
|
}
|
|
}
|
|
|
|
private static void EnsureActiveNode(IServiceProvider sp)
|
|
{
|
|
var gate = sp.GetService<IActiveNodeGate>();
|
|
if (gate is not null && !gate.IsActiveNode)
|
|
{
|
|
throw new ManagementCommandException(
|
|
"Live runs execute on the ACTIVE central node only (Component-InboundAPI: method scripts and " +
|
|
"Route.To never run on a standby). Open the Central UI or point the CLI at the active node.",
|
|
StandbyNodeCode);
|
|
}
|
|
}
|
|
|
|
private static void EnforceSiteScope(AuthenticatedUser user, int siteId)
|
|
{
|
|
if (user.PermittedSiteIds.Length == 0) return;
|
|
if (user.Roles.Contains(Roles.Administrator, StringComparer.OrdinalIgnoreCase)) return;
|
|
if (!user.PermittedSiteIds.Contains(siteId.ToString()))
|
|
{
|
|
throw new SiteScopeViolationException(
|
|
$"Access denied: your Deployment role is scoped to sites [{string.Join(", ", user.PermittedSiteIds)}] " +
|
|
$"and does not include site {siteId}.");
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Arch-review F2: the operator's <c>PermittedSiteIds</c> — NUMERIC site ids as strings —
|
|
/// resolved to the site IDENTIFIERS <see cref="IInstanceLocator"/> speaks, for
|
|
/// <c>RouteHelper.WithSiteScope</c>. Returns null (unscoped) for an unscoped user or an
|
|
/// Administrator, exactly as <see cref="EnforceSiteScope"/> treats them.
|
|
///
|
|
/// <para>
|
|
/// A returned set may be EMPTY — a scoped user whose sites no longer exist routes nowhere.
|
|
/// That is the intended fail-closed outcome, and is why null and empty are not the same.
|
|
/// </para>
|
|
/// </summary>
|
|
private static async Task<IReadOnlySet<string>?> ResolvePermittedSiteIdentifiersAsync(
|
|
IServiceProvider sp, AuthenticatedUser user, CancellationToken ct)
|
|
{
|
|
if (user.PermittedSiteIds.Length == 0) return null;
|
|
if (user.Roles.Contains(Roles.Administrator, StringComparer.OrdinalIgnoreCase)) return null;
|
|
|
|
var sites = sp.GetRequiredService<ISiteRepository>();
|
|
var set = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
|
|
foreach (var raw in user.PermittedSiteIds)
|
|
{
|
|
if (!int.TryParse(raw, out var id)) continue;
|
|
var site = await sites.GetSiteByIdAsync(id, ct).ConfigureAwait(false);
|
|
if (site is not null) set.Add(site.SiteIdentifier);
|
|
}
|
|
|
|
return set;
|
|
}
|
|
|
|
private async Task WriteCentralAsync(IServiceProvider sp, AuditEvent evt, CancellationToken ct)
|
|
{
|
|
var writer = sp.GetService<ICentralAuditWriter>();
|
|
if (writer is null) return;
|
|
try
|
|
{
|
|
await writer.WriteAsync(evt, ct).ConfigureAwait(false);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Flow Debugger central audit write failed for {Action}", evt.Action);
|
|
}
|
|
}
|
|
}
|