diff --git a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/RouteHelper.cs b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/RouteHelper.cs index f4deebb6..8a530e9a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.InboundAPI/RouteHelper.cs +++ b/src/ZB.MOM.WW.ScadaBridge.InboundAPI/RouteHelper.cs @@ -1,7 +1,13 @@ +using System.Diagnostics; +using System.Text.Json; using Akka.Actor; +using ZB.MOM.WW.Audit; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi; using ZB.MOM.WW.ScadaBridge.Commons.Types; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; namespace ZB.MOM.WW.ScadaBridge.InboundAPI; @@ -47,16 +53,27 @@ public class RouteHelper private readonly CancellationToken _deadlineToken; private readonly CancellationToken _requestAbortedToken; private readonly Guid? _parentExecutionId; + private readonly ICentralAuditWriter? _auditWriter; + private readonly INodeIdentityProvider? _nodeIdentity; + private readonly bool _trace; + private readonly Action? _traceSink; /// /// Initializes a new with no deadline token and no parent execution id. + /// DI (AddScoped<RouteHelper>()) resolves the two optional audit collaborators when the + /// host registers them; a bare new RouteHelper(locator, router) writes no audit rows. /// /// Service to resolve the site id for a given instance code. /// Service to route cross-site calls to the resolved site. + /// Flow Debugger (design 2026-09-05 §5.1): central-direct-write audit writer for the always-on RouteCall row; null = no row. + /// Stamps SourceNode (central-a/central-b) on the RouteCall row; null = NULL. public RouteHelper( IInstanceLocator instanceLocator, - IInstanceRouter instanceRouter) - : this(instanceLocator, instanceRouter, CancellationToken.None, CancellationToken.None, parentExecutionId: null) + IInstanceRouter instanceRouter, + ICentralAuditWriter? auditWriter = null, + INodeIdentityProvider? nodeIdentity = null) + : this(instanceLocator, instanceRouter, CancellationToken.None, CancellationToken.None, parentExecutionId: null, + auditWriter, nodeIdentity, trace: false, traceSink: null) { } @@ -65,13 +82,21 @@ public class RouteHelper IInstanceRouter instanceRouter, CancellationToken deadlineToken, CancellationToken requestAbortedToken, - Guid? parentExecutionId) + Guid? parentExecutionId, + ICentralAuditWriter? auditWriter, + INodeIdentityProvider? nodeIdentity, + bool trace, + Action? traceSink) { _instanceLocator = instanceLocator; _instanceRouter = instanceRouter; _deadlineToken = deadlineToken; _requestAbortedToken = requestAbortedToken; _parentExecutionId = parentExecutionId; + _auditWriter = auditWriter; + _nodeIdentity = nodeIdentity; + _trace = trace; + _traceSink = traceSink; } /// @@ -84,7 +109,8 @@ public class RouteHelper /// The executing method's timeout cancellation token to inherit for routed calls. /// A new inheriting the given deadline token. public RouteHelper WithDeadline(CancellationToken deadlineToken) => - new(_instanceLocator, _instanceRouter, deadlineToken, _requestAbortedToken, _parentExecutionId); + new(_instanceLocator, _instanceRouter, deadlineToken, _requestAbortedToken, _parentExecutionId, + _auditWriter, _nodeIdentity, _trace, _traceSink); /// /// Returns a carrying the raw request-abort @@ -97,7 +123,8 @@ public class RouteHelper /// The raw client-disconnect token, independent of the method timeout. /// A new carrying the given request-abort token. public RouteHelper WithRequestAborted(CancellationToken requestAbortedToken) => - new(_instanceLocator, _instanceRouter, _deadlineToken, requestAbortedToken, _parentExecutionId); + new(_instanceLocator, _instanceRouter, _deadlineToken, requestAbortedToken, _parentExecutionId, + _auditWriter, _nodeIdentity, _trace, _traceSink); /// /// Audit Log (ParentExecutionId): returns a whose @@ -111,7 +138,31 @@ public class RouteHelper /// The inbound request's execution id to stamp as the spawning parent on routed calls, or null for non-routed runs. /// A new carrying the given parent execution id. public RouteHelper WithParentExecutionId(Guid? parentExecutionId) => - new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, parentExecutionId); + new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, parentExecutionId, + _auditWriter, _nodeIdentity, _trace, _traceSink); + + /// + /// Flow Debugger (design 2026-09-05 §5.2): returns a whose routed + /// calls and writes carry Trace = , so the site traces the + /// spawned run. The always-on RouteCall row is written regardless of this flag. + /// + /// True to trace the routed site runs. + /// A new carrying the flag. + public RouteHelper WithTrace(bool trace) => + new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId, + _auditWriter, _nodeIdentity, trace, _traceSink); + + /// + /// Flow Debugger (design 2026-09-05 §7.1): returns a that also + /// hands every RouteCall row AND every inline site trace event on the routed reply + /// to , so a Live run can render without a round-trip to the + /// audit store. Null clears the sink. + /// + /// Receives the audit events in emission order; null = none. + /// A new carrying the sink. + public RouteHelper WithTraceSink(Action? traceSink) => + new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId, + _auditWriter, _nodeIdentity, _trace, traceSink); /// /// Creates a route target for the specified instance. @@ -121,7 +172,8 @@ public class RouteHelper public RouteTarget To(string instanceCode) { return new RouteTarget( - instanceCode, _instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId); + instanceCode, _instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId, + _auditWriter, _nodeIdentity, _trace, _traceSink); } } @@ -136,6 +188,10 @@ public class RouteTarget private readonly CancellationToken _deadlineToken; private readonly CancellationToken _requestAbortedToken; private readonly Guid? _parentExecutionId; + private readonly ICentralAuditWriter? _auditWriter; + private readonly INodeIdentityProvider? _nodeIdentity; + private readonly bool _trace; + private readonly Action? _traceSink; /// /// Initializes a new for the specified instance. @@ -152,7 +208,11 @@ public class RouteTarget IInstanceRouter instanceRouter, CancellationToken deadlineToken, CancellationToken requestAbortedToken, - Guid? parentExecutionId) + Guid? parentExecutionId, + ICentralAuditWriter? auditWriter = null, + INodeIdentityProvider? nodeIdentity = null, + bool trace = false, + Action? traceSink = null) { _instanceCode = instanceCode; _instanceLocator = instanceLocator; @@ -160,6 +220,10 @@ public class RouteTarget _deadlineToken = deadlineToken; _requestAbortedToken = requestAbortedToken; _parentExecutionId = parentExecutionId; + _auditWriter = auditWriter; + _nodeIdentity = nodeIdentity; + _trace = trace; + _traceSink = traceSink; } /// @@ -188,11 +252,31 @@ public class RouteTarget // (the inbound API request's ExecutionId) so the routed site script // records this call's parent. CorrelationId above is a separate concern // — the per-operation lifecycle id, freshly minted per routed call. + var normalized = ScriptArgs.Normalize(parameters); var request = new RouteToCallRequest( - correlationId, _instanceCode, scriptName, ScriptArgs.Normalize(parameters), - DateTimeOffset.UtcNow, _parentExecutionId); + correlationId, _instanceCode, scriptName, normalized, + DateTimeOffset.UtcNow, _parentExecutionId, + // Flow Debugger (design 2026-09-05 §5.2): the trace flag rides beside the parent id. + Trace: _trace); - var response = await RouteOrUnreachable(_instanceRouter.RouteToCallAsync(siteId, request, token)); + // Flow Debugger (design §5.1): the ALWAYS-ON RouteCall row — one per routed call, + // written whether or not the run is traced (a writer must be present). + var startedUtc = DateTime.UtcNow; + var sw = Stopwatch.StartNew(); + RouteToCallResponse response; + try + { + response = await RouteOrUnreachable(_instanceRouter.RouteToCallAsync(siteId, request, token)); + } + catch (Exception ex) + { + EmitRouteCall(scriptName, normalized, startedUtc, sw.ElapsedMilliseconds, success: false, + returnValue: null, error: ex.Message, childExecutionId: null, traceEvents: null); + throw; + } + + EmitRouteCall(scriptName, normalized, startedUtc, sw.ElapsedMilliseconds, response.Success, + response.ReturnValue, response.ErrorMessage, response.ChildExecutionId, response.TraceEvents); if (!response.Success) { @@ -203,6 +287,78 @@ public class RouteTarget return response.ReturnValue; } + /// + /// Flow Debugger (design 2026-09-05 §5.1/§7.1): builds the ScriptTrace.RouteCall row, + /// writes it best-effort through the central audit writer (never throws into the script), + /// and hands it — followed by the site's inline trace events — to the trace sink. + /// + private void EmitRouteCall( + string scriptName, IReadOnlyDictionary? args, DateTime startedUtc, long elapsedMs, + bool success, object? returnValue, string? error, Guid? childExecutionId, + IReadOnlyList? traceEvents) + { + if (_auditWriter is null && _traceSink is null) return; + AuditEvent evt; + try + { + evt = ScadaBridgeAuditEventFactory.Create( + AuditChannel.ScriptTrace, AuditKind.RouteCall, + success ? AuditStatus.Delivered : AuditStatus.Failed, + occurredAtUtc: DateTime.SpecifyKind(startedUtc, DateTimeKind.Utc), + actor: "inbound-api", + target: $"{_instanceCode}.{scriptName}", + sourceNode: _nodeIdentity?.NodeName, + executionId: _parentExecutionId, + parentExecutionId: null, + durationMs: (int)Math.Min(elapsedMs, int.MaxValue), + errorMessage: error, + requestSummary: TraceJson(args), + responseSummary: TraceJson(returnValue), + extra: childExecutionId is null ? null : new ScriptTraceExtra { ChildExecutionId = childExecutionId }.ToJson()); + } + catch (Exception) + { + return; + } + + if (_auditWriter is not null) + { + try + { + var t = _auditWriter.WriteAsync(evt, CancellationToken.None); + if (!t.IsCompletedSuccessfully) + { + _ = t.ContinueWith(static _ => { }, TaskContinuationOptions.OnlyOnFaulted); + } + } + catch (Exception) + { + // Best-effort: an audit failure never fails the routed call. + } + } + + if (_traceSink is null) return; + try + { + _traceSink(evt); + if (traceEvents is not null) + { + foreach (var e in traceEvents) _traceSink(e); + } + } + catch (Exception) + { + // A misbehaving sink must not fail the script either. + } + } + + private static string? TraceJson(object? value) + { + if (value is null) return null; + try { return JsonSerializer.Serialize(value, ScriptTraceExtra.JsonOptions); } + catch (Exception) { return null; } + } + /// /// Gets a single attribute value from the remote instance. /// @@ -353,7 +509,9 @@ public class RouteTarget // across Call / GetAttributes / SetAttributes. var request = new RouteToSetAttributesRequest( correlationId, _instanceCode, attributeValues, DateTimeOffset.UtcNow, - _parentExecutionId); + _parentExecutionId, + // Flow Debugger (design §5.3): a traced routed write arms the site's echo stamp. + Trace: _trace); var response = await RouteOrUnreachable(_instanceRouter.RouteToSetAttributesAsync(siteId, request, token)); diff --git a/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/RouteHelperTraceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/RouteHelperTraceTests.cs new file mode 100644 index 00000000..cef4fcf2 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.InboundAPI.Tests/RouteHelperTraceTests.cs @@ -0,0 +1,147 @@ +using NSubstitute; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Debugging; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.InboundAPI.Tests; + +/// +/// Flow Debugger (design 2026-09-05 §5.1/§5.2/§7.1): writes an +/// ALWAYS-ON ScriptTrace.RouteCall row (traced or not), propagates Trace on both +/// route requests, and feeds the row plus the reply's inline site events to the trace sink. +/// +public class RouteHelperTraceTests +{ + private readonly IInstanceLocator _locator = Substitute.For(); + private readonly IInstanceRouter _router = Substitute.For(); + private readonly List _written = new(); + private readonly ICentralAuditWriter _writer = Substitute.For(); + private readonly INodeIdentityProvider _node = Substitute.For(); + + public RouteHelperTraceTests() + { + _locator.GetSiteIdForInstanceAsync("inst-1", Arg.Any()).Returns("SiteA"); + _writer.WriteAsync(Arg.Do(e => _written.Add(e)), Arg.Any()).Returns(Task.CompletedTask); + _node.NodeName.Returns("central-a"); + } + + private RouteHelper Helper() => new(_locator, _router, _writer, _node); + + private static AuditEvent SiteEvent(Guid exec) => ScadaBridgeAuditEventFactory.Create( + AuditChannel.ScriptTrace, AuditKind.ScriptRun, AuditStatus.Delivered, executionId: exec, target: "doWork"); + + [Fact] + public async Task Call_WritesRouteCallRow_WithParentAsExecutionId_AndChildFromReply() + { + var parent = Guid.NewGuid(); + var child = Guid.NewGuid(); + _router.RouteToCallAsync("SiteA", Arg.Any(), Arg.Any()) + .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 99, null, + DateTimeOffset.UtcNow, child, null)); + + var result = await Helper().WithParentExecutionId(parent).To("inst-1").Call("doWork", new { x = 1 }); + + Assert.Equal(99, result); + var evt = Assert.Single(_written); + var row = evt.AsRow(); + Assert.Equal("ScriptTrace.RouteCall", evt.Action); + Assert.Equal("inbound-api", evt.Actor); + Assert.Equal("inst-1.doWork", evt.Target); + Assert.Equal("central-a", evt.SourceNode); + Assert.Equal(parent, row.ExecutionId); + Assert.Null(row.ParentExecutionId); + Assert.Equal(AuditStatus.Delivered, row.Status); + Assert.Equal(child, ScriptTraceExtra.TryParse(row.Extra)!.ChildExecutionId); + Assert.Contains("\"x\":1", row.RequestSummary); + Assert.Equal("99", row.ResponseSummary); + } + + [Fact] + public async Task Call_IsAlwaysOn_EvenWhenNotTraced_AndTraceFalseOnRequest() + { + RouteToCallRequest? captured = null; + _router.RouteToCallAsync("SiteA", Arg.Do(r => captured = r), Arg.Any()) + .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, null, null, DateTimeOffset.UtcNow)); + + await Helper().To("inst-1").Call("doWork"); + + Assert.False(captured!.Trace); + Assert.Single(_written); + } + + [Fact] + public async Task WithTrace_PropagatesOnCallAndSetAttributes() + { + RouteToCallRequest? call = null; + RouteToSetAttributesRequest? set = null; + _router.RouteToCallAsync("SiteA", Arg.Do(r => call = r), Arg.Any()) + .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, null, null, DateTimeOffset.UtcNow)); + _router.RouteToSetAttributesAsync("SiteA", Arg.Do(r => set = r), Arg.Any()) + .Returns(ci => new RouteToSetAttributesResponse(((RouteToSetAttributesRequest)ci[1]).CorrelationId, true, null, DateTimeOffset.UtcNow)); + + var traced = Helper().WithTrace(true).WithDeadline(CancellationToken.None).To("inst-1"); + await traced.Call("doWork"); + await traced.SetAttributes(new Dictionary { ["A"] = "1" }); + + Assert.True(call!.Trace); + Assert.True(set!.Trace); + } + + [Fact] + public async Task Sink_ReceivesRouteCallRow_ThenTheReplysTraceEvents_InOrder() + { + var child = Guid.NewGuid(); + var siteEvents = new[] { SiteEvent(child), SiteEvent(child) }; + _router.RouteToCallAsync("SiteA", Arg.Any(), Arg.Any()) + .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, null, null, + DateTimeOffset.UtcNow, child, siteEvents)); + var sunk = new List(); + + await Helper().WithTrace(true).WithTraceSink(sunk.Add).To("inst-1").Call("doWork"); + + Assert.Equal(3, sunk.Count); + Assert.Equal("ScriptTrace.RouteCall", sunk[0].Action); + Assert.Same(siteEvents[0], sunk[1]); + Assert.Same(siteEvents[1], sunk[2]); + } + + [Fact] + public async Task Failure_WritesFailedRow_AndStillThrows() + { + _router.RouteToCallAsync("SiteA", Arg.Any(), Arg.Any()) + .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, false, null, "script faulted", DateTimeOffset.UtcNow)); + + await Assert.ThrowsAsync(() => Helper().To("inst-1").Call("doWork")); + + var row = Assert.Single(_written).AsRow(); + Assert.Equal(AuditStatus.Failed, row.Status); + Assert.Equal("script faulted", row.ErrorMessage); + } + + [Fact] + public async Task NoWriterAndNoSink_WritesNothing_AndNeverThrows() + { + _router.RouteToCallAsync("SiteA", Arg.Any(), Arg.Any()) + .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 1, null, DateTimeOffset.UtcNow)); + + var result = await new RouteHelper(_locator, _router).WithTrace(true).To("inst-1").Call("doWork"); + + Assert.Equal(1, result); + Assert.Empty(_written); + } + + [Fact] + public async Task FaultingWriter_DoesNotFailTheCall() + { + var badWriter = Substitute.For(); + badWriter.WriteAsync(Arg.Any(), Arg.Any()).Returns(Task.FromException(new IOException("disk"))); + _router.RouteToCallAsync("SiteA", Arg.Any(), Arg.Any()) + .Returns(ci => new RouteToCallResponse(((RouteToCallRequest)ci[1]).CorrelationId, true, 5, null, DateTimeOffset.UtcNow)); + + var result = await new RouteHelper(_locator, _router, badWriter, null).To("inst-1").Call("doWork"); + Assert.Equal(5, result); + } +}