From 83aa76c33dca1a30045c1a0c2b40be10b705e3b0 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Tue, 13 Oct 2026 10:35:28 -0400 Subject: [PATCH] Retain fake worker fault diagnostics in test fixtures --- .../GatewayEndToEndMultiSubscriberTests.cs | 45 ++++-- .../GatewayEndToEndReconnectReplayTests.cs | 71 +++++++-- .../Workers/FakeWorkerDiagnosticsTests.cs | 88 ++++++++++++ .../Workers/GatedWorkerEnvelopeReaderTests.cs | 27 +++- .../TestSupport/FakeWorkerDiagnostics.cs | 135 ++++++++++++++++++ 5 files changed, 340 insertions(+), 26 deletions(-) create mode 100644 src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/FakeWorkerDiagnosticsTests.cs create mode 100644 src/ZB.MOM.WW.MxGateway.Tests/TestSupport/FakeWorkerDiagnostics.cs diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndMultiSubscriberTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndMultiSubscriberTests.cs index 4dfceed..f50c081 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndMultiSubscriberTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndMultiSubscriberTests.cs @@ -2,6 +2,7 @@ using Google.Protobuf.WellKnownTypes; using Grpc.Core; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; +using Xunit.Abstractions; using ZB.MOM.WW.MxGateway.Contracts; using ZB.MOM.WW.MxGateway.Contracts.Proto; using ZB.MOM.WW.MxGateway.Server.Configuration; @@ -21,7 +22,7 @@ namespace ZB.MOM.WW.MxGateway.Tests.Gateway; /// the fake worker harness. Covers fan-out to two concurrent subscribers, independent /// cancellation isolation, and the per-session subscriber cap. /// -public sealed class GatewayEndToEndMultiSubscriberTests +public sealed class GatewayEndToEndMultiSubscriberTests(ITestOutputHelper output) { private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(10); private const int ServerHandle = 3001; @@ -37,7 +38,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests public async Task StreamEvents_TwoSubscribers_BothReceiveAllEvents() { MultiEventFakeWorkerProcessLauncher launcher = new(EventCount); - await using MultiSubscriberGatewayServiceFixture fixture = new(launcher); + await using MultiSubscriberGatewayServiceFixture fixture = new(launcher, diagnosticOutput: output.WriteLine); OpenSessionReply openReply = await fixture.Service.OpenSession( new OpenSessionRequest @@ -136,7 +137,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests public async Task StreamEvents_OneSubscriberCancels_OtherContinuesReceivingEvents() { GatedEventFakeWorkerProcessLauncher launcher = new(); - await using MultiSubscriberGatewayServiceFixture fixture = new(launcher); + await using MultiSubscriberGatewayServiceFixture fixture = new(launcher, diagnosticOutput: output.WriteLine); OpenSessionReply openReply = await fixture.Service.OpenSession( new OpenSessionRequest @@ -254,7 +255,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests // (EventSubscriberLimitReached) is only reachable when multi-subscriber mode is on. // Without it the service hits EventSubscriberAlreadyActive instead — also // ResourceExhausted, but for the wrong reason — masking a misconfiguration. - await using MultiSubscriberGatewayServiceFixture fixture = new(launcher, maxEventSubscribersPerSession: cap); + await using MultiSubscriberGatewayServiceFixture fixture = new(launcher, maxEventSubscribersPerSession: cap, diagnosticOutput: output.WriteLine); OpenSessionReply openReply = await fixture.Service.OpenSession( new OpenSessionRequest @@ -413,20 +414,25 @@ public sealed class GatewayEndToEndMultiSubscriberTests { private readonly GatewayMetrics _metrics = new(); private readonly SessionRegistry _registry = new(); + private readonly FakeWorkerDiagnostics _diagnostics; /// Initializes a new instance of the class. /// Fake worker process launcher backing the session manager. + /// Safe metadata output sink. /// Maximum concurrent event subscribers allowed per session. public MultiSubscriberGatewayServiceFixture( IWorkerProcessLauncher launcher, - int maxEventSubscribersPerSession = 8) + int maxEventSubscribersPerSession = 8, + Action? diagnosticOutput = null) { + _diagnostics = ((IDiagnosticFakeWorker)launcher).Diagnostics; + _diagnostics.SetOutput(diagnosticOutput); IOptions options = Options.Create(CreateOptions(maxEventSubscribersPerSession)); SessionWorkerClientFactory workerClientFactory = new( launcher, options, _metrics, - NullLoggerFactory.Instance); + _diagnostics); SessionManager sessionManager = new( _registry, workerClientFactory, @@ -499,13 +505,20 @@ public sealed class GatewayEndToEndMultiSubscriberTests /// Disposes every session in the registry and releases the fixture's metrics. /// A task that represents the asynchronous operation. public async ValueTask DisposeAsync() + { + // Session disposal and worker observation share the existing test budget. + // The observer consumes a late cleanup fault without replacing the primary failure. + _diagnostics.Track(DisposeSessionsAsync(), "teardown"); + await _diagnostics.DrainAsync(TestTimeout).ConfigureAwait(false); + _metrics.Dispose(); + } + + private async Task DisposeSessionsAsync() { foreach (GatewaySession session in _registry.Snapshot()) { - await session.DisposeAsync(); + await session.DisposeAsync().ConfigureAwait(false); } - - _metrics.Dispose(); } private static GatewayOptions CreateOptions(int maxEventSubscribersPerSession) => @@ -540,7 +553,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests /// command, then waits for shutdown. Events carry an indexed string value so the test can /// verify fan-out order across two subscribers. /// - private sealed class MultiEventFakeWorkerProcessLauncher(int eventCount) : IWorkerProcessLauncher + private sealed class MultiEventFakeWorkerProcessLauncher(int eventCount) : IWorkerProcessLauncher, IDiagnosticFakeWorker { public const int ProcessId = 7710; @@ -549,12 +562,16 @@ public sealed class GatewayEndToEndMultiSubscriberTests /// Gets the task representing the fake worker's running background loop. public Task WorkerTask { get; private set; } = Task.CompletedTask; + /// + public FakeWorkerDiagnostics Diagnostics { get; } = new(); + /// public Task LaunchAsync( WorkerProcessLaunchRequest request, CancellationToken cancellationToken = default) { WorkerTask = RunWorkerAsync(request, cancellationToken); + Diagnostics.Track(WorkerTask); return Task.FromResult(new WorkerProcessHandle( _process, @@ -578,6 +595,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests { WorkerEnvelope envelope = await harness.ReadGatewayEnvelopeAsync(cancellationToken).ConfigureAwait(false); + Diagnostics.Received(envelope); if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerShutdown) { await harness.SendShutdownAckAsync(cancellationToken: cancellationToken).ConfigureAwait(false); @@ -630,7 +648,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests /// deterministic. Call before closing the session so the /// worker loop exits cleanly and can process the shutdown envelope. /// - private sealed class GatedEventFakeWorkerProcessLauncher : IWorkerProcessLauncher + private sealed class GatedEventFakeWorkerProcessLauncher : IWorkerProcessLauncher, IDiagnosticFakeWorker { public const int ProcessId = 7720; @@ -643,6 +661,9 @@ public sealed class GatewayEndToEndMultiSubscriberTests /// Gets the task representing the fake worker's running background loop. public Task WorkerTask { get; private set; } = Task.CompletedTask; + /// + public FakeWorkerDiagnostics Diagnostics { get; } = new(); + /// Releases the gate so the worker emits one event. public void AllowNextEvent() => _emitGate.Release(); @@ -662,6 +683,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests CancellationToken cancellationToken = default) { WorkerTask = RunWorkerAsync(request, cancellationToken); + Diagnostics.Track(WorkerTask); return Task.FromResult(new WorkerProcessHandle( _process, @@ -717,6 +739,7 @@ public sealed class GatewayEndToEndMultiSubscriberTests continue; } + Diagnostics.Received(envelope); if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerShutdown) { await harness.SendShutdownAckAsync(cancellationToken: cancellationToken).ConfigureAwait(false); diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs index b23eea1..fd61cb9 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs @@ -1,8 +1,10 @@ +using System.Collections.Concurrent; using Google.Protobuf.WellKnownTypes; using Grpc.Core; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using Microsoft.Extensions.Time.Testing; +using Xunit.Abstractions; using ZB.MOM.WW.MxGateway.Contracts; using ZB.MOM.WW.MxGateway.Contracts.Proto; using ZB.MOM.WW.MxGateway.Server.Configuration; @@ -36,7 +38,7 @@ namespace ZB.MOM.WW.MxGateway.Tests.Gateway; /// per session and survives detach, so events emitted while no subscriber is attached are /// retained for the reconnecting stream. /// -public sealed class GatewayEndToEndReconnectReplayTests +public sealed class GatewayEndToEndReconnectReplayTests(ITestOutputHelper output) { private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(10); private const int ServerHandle = 3001; @@ -58,7 +60,7 @@ public sealed class GatewayEndToEndReconnectReplayTests GatedEventFakeWorkerProcessLauncher launcher = new(); // Capacity 16 retains every event emitted here (7 total), so nothing is evicted and a // resume from a middle cursor never produces a gap. - await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: 16); + await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: 16, diagnosticOutput: output.WriteLine); string sessionId = await OpenSessionAsync(fixture, "reconnect-no-gap"); @@ -157,7 +159,7 @@ public sealed class GatewayEndToEndReconnectReplayTests GatedEventFakeWorkerProcessLauncher launcher = new(); // Small capacity forces eviction: with 6 events emitted and a 3-slot ring, the first 3 // are evicted and only the newest 3 remain replayable. - await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: capacity); + await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: capacity, diagnosticOutput: output.WriteLine); string sessionId = await OpenSessionAsync(fixture, "reconnect-gap"); @@ -256,7 +258,8 @@ public sealed class GatewayEndToEndReconnectReplayTests launcher, replayBufferCapacity: 16, replayRetentionSeconds: retentionSeconds, - timeProvider: clock); + timeProvider: clock, + diagnosticOutput: output.WriteLine); string sessionId = await OpenSessionAsync(fixture, "reconnect-aged-out"); @@ -340,6 +343,34 @@ public sealed class GatewayEndToEndReconnectReplayTests $"Live event {delivered.WorkerSequence} must be newer than the highest previously seen {highestSeen}."); } + /// Setup failures preserve gateway EOF and report the original fake worker fault safely. + /// The setup command where the fake worker fails. + /// The asynchronous test operation. + [Theory] + [InlineData(MxCommandKind.Register)] + [InlineData(MxCommandKind.AddItem)] + public async Task Setup_WhenFakeWorkerFaults_ReportsFaultWithoutMaskingRpcFailure(MxCommandKind faultCommand) + { + ConcurrentQueue output = new(); + GatedEventFakeWorkerProcessLauncher launcher = new(faultCommand); + Exception? failure = await Record.ExceptionAsync(async () => + { + await using ReconnectReplayGatewayServiceFixture fixture = new( + launcher, replayBufferCapacity: 16, diagnosticOutput: output.Enqueue); + string sessionId = await OpenSessionAsync(fixture, "injected-fault"); + await WireUpAdviseAsync(fixture, sessionId); + }); + + RpcException rpc = Assert.IsType(failure); + Assert.Equal(StatusCode.Unavailable, rpc.StatusCode); + Assert.Contains("Worker pipe disconnected", rpc.Status.Detail); + Assert.Contains(output, line => line.Contains("source=fake-worker") + && line.Contains("type=WorkerFrameProtocolException") && line.Contains("code=InvalidEnvelope") + && line.Contains($"command={faultCommand}") && line.Contains("phase=command")); + Assert.Contains(output, line => line.Contains("source=gateway") && line.Contains("code=EndOfStream")); + Assert.DoesNotContain(output, line => line.Contains("secret-item") || line.Contains("Galaxy.Tag.Value")); + } + // ---- shared flow helpers ---- private static async Task OpenSessionAsync( @@ -490,6 +521,7 @@ public sealed class GatewayEndToEndReconnectReplayTests { private readonly GatewayMetrics _metrics = new(); private readonly SessionRegistry _registry = new(); + private readonly FakeWorkerDiagnostics _diagnostics; /// Initializes a new instance of the class. /// Fake worker process launcher backing the session manager. @@ -499,6 +531,7 @@ public sealed class GatewayEndToEndReconnectReplayTests /// duration of a fast test; the age-eviction test shortens it and drives /// . /// + /// Safe metadata output sink. /// /// Clock handed to the session manager, which flows through to the session's event /// distributor. Pass a fake to make age-eviction deterministic. @@ -507,15 +540,18 @@ public sealed class GatewayEndToEndReconnectReplayTests IWorkerProcessLauncher launcher, int replayBufferCapacity, double replayRetentionSeconds = 300, - TimeProvider? timeProvider = null) + TimeProvider? timeProvider = null, + Action? diagnosticOutput = null) { + _diagnostics = ((IDiagnosticFakeWorker)launcher).Diagnostics; + _diagnostics.SetOutput(diagnosticOutput); IOptions options = Options.Create( CreateOptions(replayBufferCapacity, replayRetentionSeconds)); SessionWorkerClientFactory workerClientFactory = new( launcher, options, _metrics, - NullLoggerFactory.Instance); + _diagnostics); SessionManager sessionManager = new( _registry, workerClientFactory, @@ -585,13 +621,20 @@ public sealed class GatewayEndToEndReconnectReplayTests /// Disposes every session in the registry and releases the fixture's metrics. /// A task that represents the asynchronous operation. public async ValueTask DisposeAsync() + { + // Session disposal and worker observation share the existing test budget. + // The observer consumes a late cleanup fault without replacing the primary failure. + _diagnostics.Track(DisposeSessionsAsync(), "teardown"); + await _diagnostics.DrainAsync(TestTimeout).ConfigureAwait(false); + _metrics.Dispose(); + } + + private async Task DisposeSessionsAsync() { foreach (GatewaySession session in _registry.Snapshot()) { await session.DisposeAsync().ConfigureAwait(false); } - - _metrics.Dispose(); } private static GatewayOptions CreateOptions(int replayBufferCapacity, double replayRetentionSeconds) => @@ -642,7 +685,7 @@ public sealed class GatewayEndToEndReconnectReplayTests /// before closing the session so the loop exits cleanly and can /// process the shutdown envelope. /// - private sealed class GatedEventFakeWorkerProcessLauncher : IWorkerProcessLauncher + private sealed class GatedEventFakeWorkerProcessLauncher(MxCommandKind? faultCommand = null) : IWorkerProcessLauncher, IDiagnosticFakeWorker { public const int ProcessId = 7730; @@ -655,6 +698,9 @@ public sealed class GatewayEndToEndReconnectReplayTests /// Gets the task representing the fake worker's running background loop. public Task WorkerTask { get; private set; } = Task.CompletedTask; + /// + public FakeWorkerDiagnostics Diagnostics { get; } = new(); + /// Releases the gate so the worker emits one event. public void AllowNextEvent() => _emitGate.Release(); @@ -674,6 +720,7 @@ public sealed class GatewayEndToEndReconnectReplayTests CancellationToken cancellationToken = default) { WorkerTask = RunWorkerAsync(request, cancellationToken); + Diagnostics.Track(WorkerTask); return Task.FromResult(new WorkerProcessHandle( _process, @@ -729,6 +776,7 @@ public sealed class GatewayEndToEndReconnectReplayTests continue; } + Diagnostics.Received(envelope); if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerShutdown) { await harness.SendShutdownAckAsync(cancellationToken: cancellationToken).ConfigureAwait(false); @@ -742,6 +790,11 @@ public sealed class GatewayEndToEndReconnectReplayTests } MxCommand command = envelope.WorkerCommand.Command; + if (command.Kind == faultCommand) + { + throw new WorkerFrameProtocolException(WorkerFrameProtocolErrorCode.InvalidEnvelope, "secret-item"); + } + await harness.ReplyToCommandAsync( envelope, configureReply: reply => ConfigureCommandReply(reply, command.Kind), diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/FakeWorkerDiagnosticsTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/FakeWorkerDiagnosticsTests.cs new file mode 100644 index 0000000..9cfb0c8 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/FakeWorkerDiagnosticsTests.cs @@ -0,0 +1,88 @@ +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.MxGateway.Server.Workers; +using ZB.MOM.WW.MxGateway.Tests.TestSupport; + +namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Workers; + +/// Protects bounded teardown, fault observation, and safe diagnostic output. +public sealed class FakeWorkerDiagnosticsTests +{ + /// Repeated gateway logs remain bounded and never format sensitive state or messages. + [Fact] + public void Log_WhenFaultsRepeat_ReportsOnlyBoundedMetadata() + { + using FakeWorkerDiagnostics diagnostics = new(); + List output = []; + diagnostics.SetOutput(output.Add); + for (int i = 0; i < 32; i++) + { + diagnostics.Log(LogLevel.Warning, new EventId(1), "secret-command", + new WorkerFrameProtocolException(WorkerFrameProtocolErrorCode.InvalidEnvelope, "secret-item"), + (state, exception) => $"{state} {exception}"); + } + + Assert.NotEmpty(output); + Assert.InRange(output.Count, 1, 16); + Assert.All(output, line => + { + Assert.Contains("type=WorkerFrameProtocolException", line); + Assert.Contains("code=InvalidEnvelope", line); + Assert.DoesNotContain("secret", line); + }); + } + + /// A late fault is observed after timeout without calling the completed output sink. + /// The asynchronous test operation. + [Fact] + public async Task DrainAsync_WhenWorkerOutlivesBudget_ObservesLateFaultWithoutLateOutput() + { + FakeWorkerDiagnostics diagnostics = new(); + List output = []; + diagnostics.SetOutput(output.Add); + TaskCompletionSource worker = new(TaskCreationOptions.RunContinuationsAsynchronously); + diagnostics.Track(worker.Task); + await diagnostics.DrainAsync(TimeSpan.Zero); + Assert.Contains(output, line => line.Contains("source=teardown") && line.Contains("type=TimeoutException")); + int beforeFault = output.Count; + worker.SetException(new WorkerFrameProtocolException(WorkerFrameProtocolErrorCode.InvalidEnvelope, "secret-item")); + await diagnostics.DrainAsync(TimeSpan.FromSeconds(5)); + Assert.Equal(beforeFault, output.Count); + } + + /// Expected cancellation does not add fault diagnostics. + /// The asynchronous test operation. + [Fact] + public async Task DrainAsync_WhenWorkerIsCancelled_DoesNotReportFault() + { + FakeWorkerDiagnostics diagnostics = new(); + List output = []; + diagnostics.SetOutput(output.Add); + diagnostics.Track(Task.FromCanceled(new CancellationToken(true))); + await diagnostics.DrainAsync(TimeSpan.FromSeconds(5)); + Assert.Empty(output); + } + + /// An unavailable output helper cannot fault the observer or replace the primary failure. + /// The asynchronous test operation. + [Fact] + public async Task DrainAsync_WhenOutputThrows_PreservesPrimaryFailure() + { + InvalidOperationException primary = new("primary"); + FakeWorkerDiagnostics diagnostics = new(); + diagnostics.SetOutput(_ => throw new InvalidOperationException("completed output")); + Exception? failure = await Record.ExceptionAsync(async () => + { + try + { + throw primary; + } + finally + { + diagnostics.Track(Task.FromException(new WorkerFrameProtocolException( + WorkerFrameProtocolErrorCode.InvalidEnvelope, "secret-item"))); + await diagnostics.DrainAsync(TimeSpan.FromSeconds(5)); + } + }); + Assert.Same(primary, failure); + } +} diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/GatedWorkerEnvelopeReaderTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/GatedWorkerEnvelopeReaderTests.cs index 5ab097a..48de8c9 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/GatedWorkerEnvelopeReaderTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/GatedWorkerEnvelopeReaderTests.cs @@ -11,8 +11,11 @@ public sealed class GatedWorkerEnvelopeReaderTests { /// A gate wakeup must preserve a partially consumed frame and the following frame. /// The asynchronous test operation. - [Fact] - public async Task ReadOrGateAsync_WhenGateOpensDuringSegmentedFrame_PreservesBothFrames() + /// Whether the emission gate is watched (false before Advise). + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task ReadOrGateAsync_WhenGateOpensDuringSegmentedFrame_PreservesBothFrames(bool watchGate) { WorkerFrameProtocolOptions options = new("segmented-session"); WorkerEnvelope first = new() @@ -38,15 +41,27 @@ public sealed class GatedWorkerEnvelopeReaderTests // The length and first payload byte are consumed before the remaining bytes exist. await pipe.Writer.WriteAsync(bytes.AsMemory(0, 5)); - Task pending = selector.ReadOrGateAsync(); + Task pending = selector.ReadOrGateAsync(watchGate); Assert.False(pending.IsCompleted); gate.Release(); - Assert.Null(await pending); + if (watchGate) + { + Assert.Null(await pending.WaitAsync(TimeSpan.FromSeconds(5))); + } + else + { + // Explicit deadline probe: pre-Advise framing must remain pending beyond the old + // 50ms cancellation budget, even with the emission gate already open. + Task oldReadBudget = Task.Delay(TimeSpan.FromMilliseconds(100), lifetime.Token); + Assert.Same(oldReadBudget, await Task.WhenAny(pending, oldReadBudget)); + await oldReadBudget; + } + await pipe.Writer.WriteAsync(bytes.AsMemory(5)); await pipe.Writer.CompleteAsync(); - Assert.Equal(first, await selector.ReadOrGateAsync()); - Assert.Equal(second, await selector.ReadOrGateAsync()); + Assert.Equal(first, watchGate ? await selector.ReadOrGateAsync() : await pending.WaitAsync(TimeSpan.FromSeconds(5))); + Assert.Equal(second, await selector.ReadOrGateAsync(watchGate)); } /// Disposal or lifetime cancellation drains the retained read and gate wait. diff --git a/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/FakeWorkerDiagnostics.cs b/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/FakeWorkerDiagnostics.cs new file mode 100644 index 0000000..9b35ee2 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/FakeWorkerDiagnostics.cs @@ -0,0 +1,135 @@ +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.MxGateway.Server.Workers; + +namespace ZB.MOM.WW.MxGateway.Tests.TestSupport; + +/// Exposes test-owned worker tasks and safe diagnostic context to fixture cleanup. +internal interface IDiagnosticFakeWorker +{ + /// Gets the diagnostic collector for this launcher's workers. + FakeWorkerDiagnostics Diagnostics { get; } +} + +/// Observes fake worker faults without replacing the test's primary exception. +internal sealed class FakeWorkerDiagnostics : ILoggerFactory, ILogger +{ + private readonly object _sync = new(); + private readonly List _observations = []; + private Action? _output; + private int _remaining = 16; + private bool _closed; + private string _phase = "startup"; + private MxCommandKind? _command; + private ulong? _sequence; + + /// Attaches the fixture's output sink before any workers are launched. + /// Destination for safe, bounded metadata. + public void SetOutput(Action? output) => _output = output; + + /// Records only structural context, never envelopes or command arguments. + /// Incoming envelope whose safe metadata is retained. + public void Received(WorkerEnvelope envelope) + { + lock (_sync) + { + _phase = envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerCommand ? "command" : "shutdown"; + _command = envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerCommand + ? envelope.WorkerCommand.Command.Kind : null; + _sequence = envelope.Sequence; + } + } + + /// Retains an observation that consumes faults even if bounded teardown times out. + /// Background task launched by the fake worker. + /// Whether the task belongs to a fake worker or fixture teardown. + public void Track(Task worker, string source = "fake-worker") + { + Task observation = worker.ContinueWith(completed => + { + // Reading Exception observes every fault; retain no unobserved continuation fault. + if (completed.Exception is { } faults) + { + foreach (Exception fault in faults.Flatten().InnerExceptions) + { + Report(source, fault); + } + } + }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); + _observations.Add(observation); + } + + /// Drains workers under the fixture's existing time limit and closes output. + /// Existing test timeout; never extended for diagnostics. + /// The bounded cleanup operation. + public async Task DrainAsync(TimeSpan timeout) + { + try + { + await Task.WhenAll(_observations).WaitAsync(timeout).ConfigureAwait(false); + } + catch (TimeoutException exception) + { + Report("teardown", exception); + } + finally + { + Dispose(); + } + } + + private void Report(string source, Exception exception) + { + lock (_sync) + { + if (_closed || _remaining-- <= 0) + { + return; + } + + string code = exception is WorkerFrameProtocolException protocol ? protocol.ErrorCode.ToString() : "none"; + string line = $"source={source} type={exception.GetType().Name} code={code} phase={_phase} command={_command} sequence={_sequence}"; + try + { + _output?.Invoke(line); + } + catch (Exception) + { + // A completed output helper must never mask the primary test failure. + } + } + } + + /// + public ILogger CreateLogger(string categoryName) => this; + + /// + public void AddProvider(ILoggerProvider provider) => throw new NotSupportedException(); + + /// + public void Dispose() + { + lock (_sync) + { + _closed = true; + _output = null; + } + } + + /// + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + /// + public bool IsEnabled(LogLevel logLevel) => logLevel >= LogLevel.Warning; + + /// + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + { + // Never invoke formatter: state and exception messages may contain item names/secrets. + if (IsEnabled(logLevel) && exception is not null) + { + Report("gateway", exception); + } + } +} -- 2.52.0