Retain fake worker fault diagnostics in test fixtures

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.
 /// </remarks>
-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}.");
     }
 
+    /// <summary>Setup failures preserve gateway EOF and report the original fake worker fault safely.</summary>
+    /// <param name="faultCommand">The setup command where the fake worker fails.</param>
+    /// <returns>The asynchronous test operation.</returns>
+    [Theory]
+    [InlineData(MxCommandKind.Register)]
+    [InlineData(MxCommandKind.AddItem)]
+    public async Task Setup_WhenFakeWorkerFaults_ReportsFaultWithoutMaskingRpcFailure(MxCommandKind faultCommand)
+    {
+        ConcurrentQueue<string> 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<RpcException>(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<string> OpenSessionAsync(
@@ -490,6 +521,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
     {
         private readonly GatewayMetrics _metrics = new();
         private readonly SessionRegistry _registry = new();
+        private readonly FakeWorkerDiagnostics _diagnostics;
 
         /// <summary>Initializes a new instance of the <see cref="ReconnectReplayGatewayServiceFixture"/> class.</summary>
         /// <param name="launcher">Fake worker process launcher backing the session manager.</param>
@@ -499,6 +531,7 @@ public sealed class GatewayEndToEndReconnectReplayTests
         ///     duration of a fast test; the age-eviction test shortens it and drives
         ///     <paramref name="timeProvider"/>.
         /// </param>
+        /// <param name="diagnosticOutput">Safe metadata output sink.</param>
         /// <param name="timeProvider">
         ///     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<string>? diagnosticOutput = null)
         {
+            _diagnostics = ((IDiagnosticFakeWorker)launcher).Diagnostics;
+            _diagnostics.SetOutput(diagnosticOutput);
             IOptions<GatewayOptions> 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
         /// <summary>Disposes every session in the registry and releases the fixture's metrics.</summary>
         /// <returns>A task that represents the asynchronous operation.</returns>
         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
     ///     <see cref="StopEmitting"/> before closing the session so the loop exits cleanly and can
     ///     process the shutdown envelope.
     /// </summary>
-    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
         /// <summary>Gets the task representing the fake worker's running background loop.</summary>
         public Task WorkerTask { get; private set; } = Task.CompletedTask;
 
+        /// <inheritdoc />
+        public FakeWorkerDiagnostics Diagnostics { get; } = new();
+
         /// <summary>Releases the gate so the worker emits one event.</summary>
         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),