feat(management): FlowDebug run-method / run-value-change / replay commands (Deployer, active-node gated, inline trace)
Flow Debugger (design 2026-09-05 §7.1): FlowDebugRunMethodCommand / FlowDebugRunValueChangeCommand / FlowDebugReplayCommand on the per-node ManagementActor (Deployer role; the two run commands are long-running), all delegating to FlowDebugRunner. RunMethod refuses on a standby with STANDBY_NODE (ManagementCommandException now carries an optional Code that replaces COMMAND_FAILED), validates parameters through the real ParameterValidator/SchemaRefResolver (made public), synthesizes the inbound row (actor = operator, Extra.source = flow-debugger) and runs the method through InboundScriptExecutor with a trace-sink RouteHelper so the reply carries the RouteCall row + the site's inline events. RunValueChange does a traced routed write via IInstanceRouter, then polls IPullTraceEventsClient (interface moved to Commons so ManagementService can consume it) until a ScriptRun parented to the root settles, else NO_TRIGGER_WITHIN_WINDOW. Replay reads IAuditLogRepository.GetTraceEventsAsync. Claude-Session: https://claude.ai/code/session_01GSLkkHKz9JrAtmJjmQK3cP
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,317 @@
|
||||
using System.Collections.Concurrent;
|
||||
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.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.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>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"/>.
|
||||
/// </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();
|
||||
var events = new ConcurrentQueue<AuditEvent>();
|
||||
var startedAt = DateTimeOffset.UtcNow;
|
||||
var sw = Stopwatch.StartNew();
|
||||
var timeout = TimeSpan.FromSeconds(Math.Max(1, method.TimeoutSeconds));
|
||||
|
||||
var route = sp.GetRequiredService<RouteHelper>().WithTraceSink(events.Enqueue);
|
||||
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,
|
||||
$"Flow Debugger run timed out after {timeout.TotalSeconds}s + {RunGrace.TotalSeconds}s grace", "TIMEOUT");
|
||||
}
|
||||
}
|
||||
|
||||
var finishedAt = DateTimeOffset.UtcNow;
|
||||
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,
|
||||
requestSummary: requestJson,
|
||||
responseSummary: result.ResultJson,
|
||||
extra: new ScriptTraceExtra { Source = SourceTag }.ToJson());
|
||||
await WriteCentralAsync(sp, inboundRow, ct).ConfigureAwait(false);
|
||||
|
||||
var ordered = new List<AuditEvent>(events.Count + 1) { inboundRow };
|
||||
ordered.AddRange(events.OrderBy(e => e.OccurredAtUtc).ThenBy(e => e.EventId));
|
||||
|
||||
return new FlowDebugRunResult(
|
||||
root, result.Success, result.ErrorMessage, result.Success ? null : result.ErrorCode,
|
||||
ordered, Truncated: false, startedAt, finishedAt);
|
||||
}
|
||||
|
||||
/// <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);
|
||||
|
||||
var root = Guid.NewGuid();
|
||||
var startedAt = DateTimeOffset.UtcNow;
|
||||
var sw = Stopwatch.StartNew();
|
||||
// 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> { [cmd.AttributeName] = cmd.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: cmd.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 = cmd.AttributeName, value = cmd.Value }, ScriptTraceExtra.JsonOptions),
|
||||
extra: new ScriptTraceExtra
|
||||
{
|
||||
Source = SourceTag,
|
||||
Writes = [new AttributeWriteDetail(cmd.AttributeName, cmd.Value, write.Success, write.ErrorMessage)],
|
||||
}.ToJson());
|
||||
await WriteCentralAsync(sp, rootRow, ct).ConfigureAwait(false);
|
||||
|
||||
if (!write.Success)
|
||||
{
|
||||
return new FlowDebugRunResult(root, false, write.ErrorMessage ?? "Attribute write failed", "WRITE_FAILED",
|
||||
[rootRow], false, startedAt, DateTimeOffset.UtcNow);
|
||||
}
|
||||
|
||||
// Poll the site for the traced run(s) the write spawned — matched by ParentExecutionId
|
||||
// == root, then their children — until a ScriptRun parented to root exists AND the
|
||||
// set has been quiet for SettleWindow, or WaitSeconds elapses.
|
||||
var puller = sp.GetRequiredService<IPullTraceEventsClient>();
|
||||
var deadline = _now() + TimeSpan.FromSeconds(Math.Max(1, cmd.WaitSeconds));
|
||||
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(e =>
|
||||
{
|
||||
var d = AuditDetailsCodec.Deserialize(e.DetailsJson);
|
||||
return d.Kind == nameof(AuditKind.ScriptRun) && d.ParentExecutionId == root;
|
||||
});
|
||||
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> { rootRow };
|
||||
ordered.AddRange(known.Values.OrderBy(e => e.OccurredAtUtc).ThenBy(e => e.EventId));
|
||||
var completed = known.Values.Any(e =>
|
||||
{
|
||||
var d = AuditDetailsCodec.Deserialize(e.DetailsJson);
|
||||
return d.Kind == nameof(AuditKind.ScriptRun) && d.ParentExecutionId == root;
|
||||
});
|
||||
return new FlowDebugRunResult(
|
||||
root, completed,
|
||||
completed ? null : $"No traced run was spawned within {cmd.WaitSeconds}s (no value-change trigger on '{cmd.AttributeName}', 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);
|
||||
}
|
||||
|
||||
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}.");
|
||||
}
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user