fix(audit): trace/outbound fault continuations run on the default scheduler and central emitters log (arch-review F8)

Claude-Session: https://claude.ai/code/session_01NcMG6xXdsJbmD2wYAgkK29
This commit is contained in:
Joseph Doherty
2026-09-06 14:28:44 -04:00
parent b8752cc855
commit 262de38cee
6 changed files with 93 additions and 18 deletions
@@ -2,6 +2,7 @@ using System.Data.Common;
using System.Diagnostics; using System.Diagnostics;
using System.Globalization; using System.Globalization;
using System.Text; using System.Text;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.Audit; using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
@@ -70,6 +71,7 @@ public sealed class InboundDatabaseHelper
private readonly Guid? _parentExecutionId; private readonly Guid? _parentExecutionId;
private readonly string? _actor; private readonly string? _actor;
private readonly Action<AuditEvent>? _traceSink; private readonly Action<AuditEvent>? _traceSink;
private readonly ILogger? _logger;
/// <summary> /// <summary>
/// Initializes the helper. /// Initializes the helper.
@@ -83,6 +85,7 @@ public sealed class InboundDatabaseHelper
/// <param name="parentExecutionId">The spawning execution's id, if any (an inbound request is a root, so callers normally pass null).</param> /// <param name="parentExecutionId">The spawning execution's id, if any (an inbound request is a root, so callers normally pass null).</param>
/// <param name="methodName">The executing API method's name; the row's <c>Actor</c> becomes <c>api:{methodName}</c>. Null = NULL actor.</param> /// <param name="methodName">The executing API method's name; the row's <c>Actor</c> becomes <c>api:{methodName}</c>. Null = NULL actor.</param>
/// <param name="traceSink">Flow Debugger trace sink; receives the same event instance before the writer does. Null = none.</param> /// <param name="traceSink">Flow Debugger trace sink; receives the same event instance before the writer does. Null = none.</param>
/// <param name="logger">Arch-review F8: logs a Warning when the best-effort audit write faults; null = silent.</param>
// Trailing optional parameters: the original (gateway, ct, commandTimeout) shape stays // Trailing optional parameters: the original (gateway, ct, commandTimeout) shape stays
// source-compatible for every existing caller and test. // source-compatible for every existing caller and test.
public InboundDatabaseHelper( public InboundDatabaseHelper(
@@ -94,7 +97,8 @@ public sealed class InboundDatabaseHelper
Guid? executionId = null, Guid? executionId = null,
Guid? parentExecutionId = null, Guid? parentExecutionId = null,
string? methodName = null, string? methodName = null,
Action<AuditEvent>? traceSink = null) Action<AuditEvent>? traceSink = null,
ILogger? logger = null)
{ {
_gateway = gateway; _gateway = gateway;
_ct = ct; _ct = ct;
@@ -107,6 +111,7 @@ public sealed class InboundDatabaseHelper
_parentExecutionId = parentExecutionId; _parentExecutionId = parentExecutionId;
_actor = string.IsNullOrEmpty(methodName) ? null : "api:" + methodName; _actor = string.IsNullOrEmpty(methodName) ? null : "api:" + methodName;
_traceSink = traceSink; _traceSink = traceSink;
_logger = logger;
} }
/// <summary>First column of the first row converted to <typeparamref name="T"/> (default if no rows).</summary> /// <summary>First column of the first row converted to <typeparamref name="T"/> (default if no rows).</summary>
@@ -268,7 +273,11 @@ public sealed class InboundDatabaseHelper
var t = _auditWriter.WriteAsync(evt, CancellationToken.None); var t = _auditWriter.WriteAsync(evt, CancellationToken.None);
if (!t.IsCompletedSuccessfully) if (!t.IsCompletedSuccessfully)
{ {
_ = t.ContinueWith(static _ => { }, TaskContinuationOptions.OnlyOnFaulted); _ = t.ContinueWith(
x => _logger?.LogWarning(x.Exception, "Flow trace DbOutbound write failed for {Connection}", connectionName),
CancellationToken.None,
TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default);
} }
} }
catch (Exception) catch (Exception)
@@ -399,7 +399,8 @@ public class InboundScriptExecutor
executionId: parentExecutionId, executionId: parentExecutionId,
parentExecutionId: null, parentExecutionId: null,
methodName: method.Name, methodName: method.Name,
traceSink: route.TraceSink); traceSink: route.TraceSink,
logger: _logger);
// Bind the route helper to the method deadline so a // Bind the route helper to the method deadline so a
// routed Route.To(...).Call(...) inherits the method-level timeout // routed Route.To(...).Call(...) inherits the method-level timeout
@@ -1,6 +1,7 @@
using System.Diagnostics; using System.Diagnostics;
using System.Text.Json; using System.Text.Json;
using Akka.Actor; using Akka.Actor;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.Audit; using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi; using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi;
@@ -87,6 +88,7 @@ public class RouteHelper
private readonly Action<AuditEvent>? _traceSink; private readonly Action<AuditEvent>? _traceSink;
private readonly IReadOnlySet<string>? _permittedSiteIdentifiers; private readonly IReadOnlySet<string>? _permittedSiteIdentifiers;
private readonly bool _captureBodies; private readonly bool _captureBodies;
private readonly ILogger? _logger;
/// <summary> /// <summary>
/// Initializes a new <see cref="RouteHelper"/> with no deadline token and no parent execution id. /// Initializes a new <see cref="RouteHelper"/> with no deadline token and no parent execution id.
@@ -97,13 +99,15 @@ public class RouteHelper
/// <param name="instanceRouter">Service to route cross-site calls to the resolved site.</param> /// <param name="instanceRouter">Service to route cross-site calls to the resolved site.</param>
/// <param name="auditWriter">Flow Debugger (design 2026-09-05 §5.1): central-direct-write audit writer for the always-on <c>RouteCall</c> row; null = no row.</param> /// <param name="auditWriter">Flow Debugger (design 2026-09-05 §5.1): central-direct-write audit writer for the always-on <c>RouteCall</c> row; null = no row.</param>
/// <param name="nodeIdentity">Stamps <c>SourceNode</c> (central-a/central-b) on the <c>RouteCall</c> row; null = NULL.</param> /// <param name="nodeIdentity">Stamps <c>SourceNode</c> (central-a/central-b) on the <c>RouteCall</c> row; null = NULL.</param>
/// <param name="logger">Arch-review F8: logs a Warning when the best-effort <c>RouteCall</c> audit write faults; null = silent.</param>
public RouteHelper( public RouteHelper(
IInstanceLocator instanceLocator, IInstanceLocator instanceLocator,
IInstanceRouter instanceRouter, IInstanceRouter instanceRouter,
ICentralAuditWriter? auditWriter = null, ICentralAuditWriter? auditWriter = null,
INodeIdentityProvider? nodeIdentity = null) INodeIdentityProvider? nodeIdentity = null,
ILogger<RouteHelper>? logger = null)
: this(instanceLocator, instanceRouter, CancellationToken.None, CancellationToken.None, parentExecutionId: null, : this(instanceLocator, instanceRouter, CancellationToken.None, CancellationToken.None, parentExecutionId: null,
auditWriter, nodeIdentity, trace: false, traceSink: null, permittedSiteIdentifiers: null, captureBodies: true) auditWriter, nodeIdentity, trace: false, traceSink: null, permittedSiteIdentifiers: null, captureBodies: true, logger)
{ {
} }
@@ -118,7 +122,8 @@ public class RouteHelper
bool trace, bool trace,
Action<AuditEvent>? traceSink, Action<AuditEvent>? traceSink,
IReadOnlySet<string>? permittedSiteIdentifiers, IReadOnlySet<string>? permittedSiteIdentifiers,
bool captureBodies) bool captureBodies,
ILogger? logger)
{ {
_permittedSiteIdentifiers = permittedSiteIdentifiers; _permittedSiteIdentifiers = permittedSiteIdentifiers;
_instanceLocator = instanceLocator; _instanceLocator = instanceLocator;
@@ -131,6 +136,7 @@ public class RouteHelper
_trace = trace; _trace = trace;
_traceSink = traceSink; _traceSink = traceSink;
_captureBodies = captureBodies; _captureBodies = captureBodies;
_logger = logger;
} }
/// <summary> /// <summary>
@@ -144,7 +150,7 @@ public class RouteHelper
/// <returns>A new <see cref="RouteHelper"/> inheriting the given deadline token.</returns> /// <returns>A new <see cref="RouteHelper"/> inheriting the given deadline token.</returns>
public RouteHelper WithDeadline(CancellationToken deadlineToken) => public RouteHelper WithDeadline(CancellationToken deadlineToken) =>
new(_instanceLocator, _instanceRouter, deadlineToken, _requestAbortedToken, _parentExecutionId, new(_instanceLocator, _instanceRouter, deadlineToken, _requestAbortedToken, _parentExecutionId,
_auditWriter, _nodeIdentity, _trace, _traceSink, _permittedSiteIdentifiers, _captureBodies); _auditWriter, _nodeIdentity, _trace, _traceSink, _permittedSiteIdentifiers, _captureBodies, _logger);
/// <summary> /// <summary>
/// Returns a <see cref="RouteHelper"/> carrying the raw request-abort /// Returns a <see cref="RouteHelper"/> carrying the raw request-abort
@@ -158,7 +164,7 @@ public class RouteHelper
/// <returns>A new <see cref="RouteHelper"/> carrying the given request-abort token.</returns> /// <returns>A new <see cref="RouteHelper"/> carrying the given request-abort token.</returns>
public RouteHelper WithRequestAborted(CancellationToken requestAbortedToken) => public RouteHelper WithRequestAborted(CancellationToken requestAbortedToken) =>
new(_instanceLocator, _instanceRouter, _deadlineToken, requestAbortedToken, _parentExecutionId, new(_instanceLocator, _instanceRouter, _deadlineToken, requestAbortedToken, _parentExecutionId,
_auditWriter, _nodeIdentity, _trace, _traceSink, _permittedSiteIdentifiers, _captureBodies); _auditWriter, _nodeIdentity, _trace, _traceSink, _permittedSiteIdentifiers, _captureBodies, _logger);
/// <summary> /// <summary>
/// Audit Log (ParentExecutionId): returns a <see cref="RouteHelper"/> whose /// Audit Log (ParentExecutionId): returns a <see cref="RouteHelper"/> whose
@@ -173,7 +179,7 @@ public class RouteHelper
/// <returns>A new <see cref="RouteHelper"/> carrying the given parent execution id.</returns> /// <returns>A new <see cref="RouteHelper"/> carrying the given parent execution id.</returns>
public RouteHelper WithParentExecutionId(Guid? parentExecutionId) => public RouteHelper WithParentExecutionId(Guid? parentExecutionId) =>
new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, parentExecutionId, new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, parentExecutionId,
_auditWriter, _nodeIdentity, _trace, _traceSink, _permittedSiteIdentifiers, _captureBodies); _auditWriter, _nodeIdentity, _trace, _traceSink, _permittedSiteIdentifiers, _captureBodies, _logger);
/// <summary> /// <summary>
/// Flow Debugger (design 2026-09-05 §5.2): returns a <see cref="RouteHelper"/> whose routed /// Flow Debugger (design 2026-09-05 §5.2): returns a <see cref="RouteHelper"/> whose routed
@@ -184,7 +190,7 @@ public class RouteHelper
/// <returns>A new <see cref="RouteHelper"/> carrying the flag.</returns> /// <returns>A new <see cref="RouteHelper"/> carrying the flag.</returns>
public RouteHelper WithTrace(bool trace) => public RouteHelper WithTrace(bool trace) =>
new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId, new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId,
_auditWriter, _nodeIdentity, trace, _traceSink, _permittedSiteIdentifiers, _captureBodies); _auditWriter, _nodeIdentity, trace, _traceSink, _permittedSiteIdentifiers, _captureBodies, _logger);
/// <summary> /// <summary>
/// Flow Debugger (design 2026-09-05 §7.1): returns a <see cref="RouteHelper"/> that also /// Flow Debugger (design 2026-09-05 §7.1): returns a <see cref="RouteHelper"/> that also
@@ -196,7 +202,7 @@ public class RouteHelper
/// <returns>A new <see cref="RouteHelper"/> carrying the sink.</returns> /// <returns>A new <see cref="RouteHelper"/> carrying the sink.</returns>
public RouteHelper WithTraceSink(Action<AuditEvent>? traceSink) => public RouteHelper WithTraceSink(Action<AuditEvent>? traceSink) =>
new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId, new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId,
_auditWriter, _nodeIdentity, _trace, traceSink, _permittedSiteIdentifiers, _captureBodies); _auditWriter, _nodeIdentity, _trace, traceSink, _permittedSiteIdentifiers, _captureBodies, _logger);
/// <summary> /// <summary>
/// Flow Debugger outbound calls (plan 2026-09-06, D5/D7): the trace sink this helper carries, /// Flow Debugger outbound calls (plan 2026-09-06, D5/D7): the trace sink this helper carries,
@@ -230,7 +236,7 @@ public class RouteHelper
/// <returns>A new <see cref="RouteHelper"/> carrying the scope.</returns> /// <returns>A new <see cref="RouteHelper"/> carrying the scope.</returns>
public RouteHelper WithSiteScope(IReadOnlySet<string>? permittedSiteIdentifiers) => public RouteHelper WithSiteScope(IReadOnlySet<string>? permittedSiteIdentifiers) =>
new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId, new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId,
_auditWriter, _nodeIdentity, _trace, _traceSink, permittedSiteIdentifiers, _captureBodies); _auditWriter, _nodeIdentity, _trace, _traceSink, permittedSiteIdentifiers, _captureBodies, _logger);
/// <summary> /// <summary>
/// Arch-review F11: returns a <see cref="RouteHelper"/> whose always-on <c>RouteCall</c> row /// Arch-review F11: returns a <see cref="RouteHelper"/> whose always-on <c>RouteCall</c> row
@@ -245,7 +251,7 @@ public class RouteHelper
/// <returns>A new <see cref="RouteHelper"/> carrying the flag.</returns> /// <returns>A new <see cref="RouteHelper"/> carrying the flag.</returns>
public RouteHelper WithBodyCapture(bool captureBodies) => public RouteHelper WithBodyCapture(bool captureBodies) =>
new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId, new(_instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId,
_auditWriter, _nodeIdentity, _trace, _traceSink, _permittedSiteIdentifiers, captureBodies); _auditWriter, _nodeIdentity, _trace, _traceSink, _permittedSiteIdentifiers, captureBodies, _logger);
/// <summary> /// <summary>
/// Creates a route target for the specified instance. /// Creates a route target for the specified instance.
@@ -256,7 +262,7 @@ public class RouteHelper
{ {
return new RouteTarget( return new RouteTarget(
instanceCode, _instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId, instanceCode, _instanceLocator, _instanceRouter, _deadlineToken, _requestAbortedToken, _parentExecutionId,
_auditWriter, _nodeIdentity, _trace, _traceSink, _permittedSiteIdentifiers, _captureBodies); _auditWriter, _nodeIdentity, _trace, _traceSink, _permittedSiteIdentifiers, _captureBodies, _logger);
} }
} }
@@ -277,6 +283,7 @@ public class RouteTarget
private readonly Action<AuditEvent>? _traceSink; private readonly Action<AuditEvent>? _traceSink;
private readonly IReadOnlySet<string>? _permittedSiteIdentifiers; private readonly IReadOnlySet<string>? _permittedSiteIdentifiers;
private readonly bool _captureBodies; private readonly bool _captureBodies;
private readonly ILogger? _logger;
/// <summary> /// <summary>
/// Initializes a new <see cref="RouteTarget"/> for the specified instance. /// Initializes a new <see cref="RouteTarget"/> for the specified instance.
@@ -293,6 +300,7 @@ public class RouteTarget
/// <param name="traceSink">Receives the emitted audit events in order; null = none.</param> /// <param name="traceSink">Receives the emitted audit events in order; null = none.</param>
/// <param name="permittedSiteIdentifiers">Arch-review F2: site identifiers this call may reach; null = unscoped.</param> /// <param name="permittedSiteIdentifiers">Arch-review F2: site identifiers this call may reach; null = unscoped.</param>
/// <param name="captureBodies">Arch-review F11: false omits the request/response summary on the <c>RouteCall</c> row.</param> /// <param name="captureBodies">Arch-review F11: false omits the request/response summary on the <c>RouteCall</c> row.</param>
/// <param name="logger">Arch-review F8: logs a Warning when the best-effort <c>RouteCall</c> audit write faults; null = silent.</param>
internal RouteTarget( internal RouteTarget(
string instanceCode, string instanceCode,
IInstanceLocator instanceLocator, IInstanceLocator instanceLocator,
@@ -305,7 +313,8 @@ public class RouteTarget
bool trace = false, bool trace = false,
Action<AuditEvent>? traceSink = null, Action<AuditEvent>? traceSink = null,
IReadOnlySet<string>? permittedSiteIdentifiers = null, IReadOnlySet<string>? permittedSiteIdentifiers = null,
bool captureBodies = true) bool captureBodies = true,
ILogger? logger = null)
{ {
_permittedSiteIdentifiers = permittedSiteIdentifiers; _permittedSiteIdentifiers = permittedSiteIdentifiers;
_instanceCode = instanceCode; _instanceCode = instanceCode;
@@ -319,6 +328,7 @@ public class RouteTarget
_trace = trace; _trace = trace;
_traceSink = traceSink; _traceSink = traceSink;
_captureBodies = captureBodies; _captureBodies = captureBodies;
_logger = logger;
} }
/// <summary> /// <summary>
@@ -425,7 +435,11 @@ public class RouteTarget
var t = _auditWriter.WriteAsync(evt, CancellationToken.None); var t = _auditWriter.WriteAsync(evt, CancellationToken.None);
if (!t.IsCompletedSuccessfully) if (!t.IsCompletedSuccessfully)
{ {
_ = t.ContinueWith(static _ => { }, TaskContinuationOptions.OnlyOnFaulted); _ = t.ContinueWith(
x => _logger?.LogWarning(x.Exception, "Flow trace RouteCall write failed for {Target}", $"{_instanceCode}.{scriptName}"),
CancellationToken.None,
TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default);
} }
} }
catch (Exception) catch (Exception)
@@ -596,7 +596,9 @@ internal static class ScriptRunLauncher
{ {
_ = t.ContinueWith( _ = t.ContinueWith(
x => logger.LogWarning(x.Exception, "Flow trace ScriptRun write failed for {Script}", scriptName), x => logger.LogWarning(x.Exception, "Flow trace ScriptRun write failed for {Script}", scriptName),
TaskContinuationOptions.OnlyOnFaulted); CancellationToken.None,
TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default);
} }
} }
catch (Exception ex) catch (Exception ex)
@@ -445,7 +445,9 @@ public class ScriptRuntimeContext
{ {
_ = t.ContinueWith( _ = t.ContinueWith(
x => _logger.LogWarning(x.Exception, "Flow trace write failed for {Kind}", kind), x => _logger.LogWarning(x.Exception, "Flow trace write failed for {Kind}", kind),
TaskContinuationOptions.OnlyOnFaulted); CancellationToken.None,
TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default);
} }
} }
catch (Exception ex) catch (Exception ex)
@@ -1,6 +1,7 @@
using System.Data.Common; using System.Data.Common;
using Microsoft.Data.Sqlite; using Microsoft.Data.Sqlite;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Logging.Abstractions;
using NSubstitute; using NSubstitute;
using ZB.MOM.WW.Audit; using ZB.MOM.WW.Audit;
@@ -58,6 +59,26 @@ public class InboundDatabaseHelperAuditTests
=> throw new NotImplementedException(); => throw new NotImplementedException();
} }
/// <summary>
/// Arch-review F8: a minimal <see cref="ILogger"/> double that records every
/// Warning-level message, mirroring the <c>CountingLogger&lt;T&gt;</c> pattern in
/// <c>InboundScriptExecutorTests</c>.
/// </summary>
private sealed class RecordingLogger : ILogger
{
public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get; } = new();
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
public bool IsEnabled(LogLevel logLevel) => true;
public void Log<TState>(
LogLevel logLevel, EventId eventId, TState state, Exception? exception,
Func<TState, Exception?, string> formatter)
{
Entries.Add((logLevel, formatter(state, exception), exception));
}
}
private static SqliteGateway SeededGateway() => new(); private static SqliteGateway SeededGateway() => new();
private readonly List<AuditEvent> _written = new(); private readonly List<AuditEvent> _written = new();
@@ -166,6 +187,32 @@ public class InboundDatabaseHelperAuditTests
Assert.Equal(1, n2); Assert.Equal(1, n2);
} }
[Fact]
public async Task AuditWriterFaulting_LogsWarning_AndCallStillSucceeds()
{
// Arch-review F8: the best-effort continuation used to be `static _ => { }` — a
// faulting write vanished with no trace anywhere. It now logs a Warning (and runs
// on TaskScheduler.Default rather than riding whatever context completed the
// audit-writer task), while the call itself still succeeds either way.
var faultedTask = Substitute.For<ICentralAuditWriter>();
faultedTask.WriteAsync(Arg.Any<AuditEvent>(), Arg.Any<CancellationToken>())
.Returns(Task.FromException(new InvalidOperationException("audit down")));
var logger = new RecordingLogger();
var helper = new InboundDatabaseHelper(
SeededGateway(), CancellationToken.None, TimeSpan.Zero,
auditWriter: faultedTask, nodeIdentity: _node,
executionId: _exec, parentExecutionId: _parent, methodName: "movein",
traceSink: null, logger: logger);
var n = await helper.ExecuteAsync("MesDb", "UPDATE Machine SET Code=Code", null);
Assert.Equal(1, n);
var entry = Assert.Single(logger.Entries);
Assert.Equal(LogLevel.Warning, entry.Level);
Assert.NotNull(entry.Exception);
Assert.Contains("MesDb", entry.Message);
}
[Fact] [Fact]
public async Task TraceSink_ReceivesTheSameInstance_BeforeTheWriter() public async Task TraceSink_ReceivesTheSameInstance_BeforeTheWriter()
{ {