From fed0685633a8c9657d9b482b4f15f1cd52542f22 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 16:26:40 -0400 Subject: [PATCH] test(gateway): TST-01 end-to-end reconnect/replay integration test Proves the default-on reconnect protocol end to end through the real gRPC StreamEvents path via the fake worker harness (no live COM). Two facts: - ReconnectInsideRetainedWindow_ReplaysTailNoGap: capacity 16 retains all events; reconnect from a mid-batch after_worker_sequence cursor replays exactly the events with WorkerSequence > cursor (retained tail + events emitted while detached), strictly ascending, distinct, no ReplayGap. - ReconnectWithStaleCursor_EmitsReplayGapSentinelFirst: capacity 3 with 6 events forces eviction; reconnect with AfterWorkerSequence=1 yields the ReplayGap sentinel first (Family=Unspecified, no body, RequestedAfterSequence=1, OldestAvailableSequence=oldest retained), then the retained tail ascending. Single-subscriber mode: the first stream is fully detached (cancel + await the stream task runs EventStreamService's finally, dropping the subscriber count to 0) before reconnect; the distributor + replay ring are created once per session and survive detach, so events emitted while detached are retained. macOS note: fake-worker E2E tests need TMPDIR=/tmp (default macOS TMPDIR pushes the CoreFxPipe Unix-socket path past the 104-byte sun_path limit). Windows CI is unaffected. Server reconnect/replay behavior matched the contract exactly. Claude-Session: https://claude.ai/code/session_01DMXXvNuPekkkrTEyPNxEkW --- .../GatewayEndToEndReconnectReplayTests.cs | 642 ++++++++++++++++++ 1 file changed, 642 insertions(+) create mode 100644 src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs new file mode 100644 index 0000000..424bd66 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs @@ -0,0 +1,642 @@ +using Google.Protobuf.WellKnownTypes; +using Grpc.Core; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.MxGateway.Contracts; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.MxGateway.Server.Configuration; +using ZB.MOM.WW.MxGateway.Server.Grpc; +using ZB.MOM.WW.MxGateway.Server.Metrics; +using ZB.MOM.WW.MxGateway.Server.Security.Authentication; +using ZB.MOM.WW.MxGateway.Server.Security.Authorization; +using ZB.MOM.WW.MxGateway.Server.Sessions; +using ZB.MOM.WW.MxGateway.Server.Workers; +using ZB.MOM.WW.MxGateway.Tests.Gateway.Workers.Fakes; +using ZB.MOM.WW.MxGateway.Tests.TestSupport; + +namespace ZB.MOM.WW.MxGateway.Tests.Gateway; + +/// +/// End-to-end reconnect/replay tests through the real gRPC StreamEvents path via the +/// fake worker harness (TST-01, server half). A single subscriber detaches (cancel + await +/// the stream task so its lease is fully disposed), the session is retained by detach-grace +/// while the distributor pump keeps appending worker events to the replay ring, and a second +/// stream reconnects with . Covers both +/// the no-gap resume (cursor inside the retained window) and the ReplayGap sentinel +/// (cursor predates the oldest retained event after capacity eviction). +/// +/// +/// These tests run in single-subscriber mode (AllowMultipleEventSubscribers=false). +/// Reconnect works because the first stream is FULLY detached before the reconnect attaches: +/// awaiting the first stream task runs EventStreamService's finally block, which +/// disposes the subscriber lease and drops the session's active-subscriber count back to +/// zero, so the reconnect's AttachEventSubscriberWithReplay sees an empty slot rather +/// than an "already active" rejection. The distributor (and its replay ring) is created once +/// per session and survives detach, so events emitted while no subscriber is attached are +/// retained for the reconnecting stream. +/// +public sealed class GatewayEndToEndReconnectReplayTests +{ + private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(10); + private const int ServerHandle = 3001; + private const int ItemHandle = 4002; + + /// + /// Reconnecting inside the retained window replays exactly the events newer than the + /// resume cursor — the retained tail of the first batch plus the events emitted while + /// detached — in strictly ascending order, with no duplicates and no ReplayGap + /// sentinel. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task StreamEvents_ReconnectInsideRetainedWindow_ReplaysTailNoGap() + { + const int firstBatch = 4; + const int secondBatch = 3; + + 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); + + string sessionId = await OpenSessionAsync(fixture, "reconnect-no-gap"); + + // ---- first connection: receive the first batch, capture its real sequences ---- + using CancellationTokenSource writer1Cts = new(); + RecordingServerStreamWriter writer1 = new(); + Task stream1Task = Task.Run(async () => + await fixture.Service.StreamEvents( + new StreamEventsRequest { SessionId = sessionId }, + writer1, + new TestServerCallContext(cancellationToken: writer1Cts.Token))); + + await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout); + await WireUpAdviseAsync(fixture, sessionId); + + for (int i = 0; i < firstBatch; i++) + { + launcher.AllowNextEvent(); + } + + IReadOnlyList batch1 = await writer1.WaitForMessageCountAsync(firstBatch, TestTimeout); + ulong[] batch1Sequences = batch1.Select(e => e.WorkerSequence).ToArray(); + + // Resume cursor: a middle event of the first batch. Everything strictly newer than this + // must be redelivered to the reconnecting stream. + ulong cursor = batch1Sequences[1]; + int retainedTailFromBatch1 = batch1Sequences.Count(s => s > cursor); + Assert.Equal(2, retainedTailFromBatch1); // sanity: events at index 2 and 3 + + // ---- fully detach writer1 (cancel + await so its lease is disposed) ---- + await DetachAsync(writer1Cts, stream1Task); + + // ---- emit more events while detached; the ring must retain them for the reconnect ---- + for (int i = 0; i < secondBatch; i++) + { + launcher.AllowNextEvent(); + } + + // ---- reconnect with the middle cursor ---- + RecordingServerStreamWriter writer2 = new(); + Task stream2Task = Task.Run(async () => + await fixture.Service.StreamEvents( + new StreamEventsRequest { SessionId = sessionId, AfterWorkerSequence = cursor }, + writer2, + new TestServerCallContext())); + + await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout); + + int expected = retainedTailFromBatch1 + secondBatch; + IReadOnlyList resumed = await writer2.WaitForMessageCountAsync(expected, TestTimeout); + + // ---- tear down before asserting so a hang can't wedge the run ---- + launcher.StopEmitting(); + await CloseAndDrainAsync(fixture, sessionId, stream2Task, launcher); + + // ---- assertions ---- + Assert.Equal(expected, resumed.Count); + + // No sentinel: every message is a real event, none carries a ReplayGap. + Assert.All(resumed, e => Assert.Null(e.ReplayGap)); + + // Every replayed/live event is strictly newer than the cursor. + Assert.All(resumed, e => Assert.True( + e.WorkerSequence > cursor, + $"Event sequence {e.WorkerSequence} is not newer than cursor {cursor}.")); + + // Strictly ascending, no duplicates. + for (int i = 1; i < resumed.Count; i++) + { + Assert.True( + resumed[i].WorkerSequence > resumed[i - 1].WorkerSequence, + $"Sequences must be strictly ascending: {resumed[i - 1].WorkerSequence} then {resumed[i].WorkerSequence}."); + } + + Assert.Equal(resumed.Count, resumed.Select(e => e.WorkerSequence).Distinct().Count()); + + // The retained tail of the first batch (the two events newer than the cursor) is + // replayed first, before the events emitted while detached. + Assert.Equal(batch1Sequences[2], resumed[0].WorkerSequence); + Assert.Equal(batch1Sequences[3], resumed[1].WorkerSequence); + } + + /// + /// Reconnecting with a cursor that predates the oldest retained event (after capacity + /// eviction) yields the ReplayGap sentinel FIRST — family unspecified, no body, no + /// per-item fields, correct requested/oldest sequences — followed by exactly the retained + /// tail, in ascending order. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task StreamEvents_ReconnectWithStaleCursor_EmitsReplayGapSentinelFirst() + { + const int capacity = 3; + const int totalEvents = 6; + + 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); + + string sessionId = await OpenSessionAsync(fixture, "reconnect-gap"); + + using CancellationTokenSource writer1Cts = new(); + RecordingServerStreamWriter writer1 = new(); + Task stream1Task = Task.Run(async () => + await fixture.Service.StreamEvents( + new StreamEventsRequest { SessionId = sessionId }, + writer1, + new TestServerCallContext(cancellationToken: writer1Cts.Token))); + + await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout); + await WireUpAdviseAsync(fixture, sessionId); + + // Emit more events than the ring can hold. Waiting for writer1 to receive all of them + // proves the pump has appended (and evicted) every event, so the ring is settled to its + // final newest-`capacity` contents before we reconnect. + for (int i = 0; i < totalEvents; i++) + { + launcher.AllowNextEvent(); + } + + IReadOnlyList all = await writer1.WaitForMessageCountAsync(totalEvents, TestTimeout); + ulong[] sequences = all.Select(e => e.WorkerSequence).ToArray(); + + // With capacity C and N total events, the newest C are retained; the oldest still + // available is the (N-C+1)th event — index N-C in emission (ascending) order. + ulong oldestAvailable = sequences[totalEvents - capacity]; + + await DetachAsync(writer1Cts, stream1Task); + + // Resume after sequence 1: the real event sequences are envelope-numbered and start well + // above 1 (startup + command-reply envelopes consume the low numbers), so cursor 1 + // predates every event and, with the oldest three already evicted, forces a gap. + const ulong staleCursor = 1; + RecordingServerStreamWriter writer2 = new(); + Task stream2Task = Task.Run(async () => + await fixture.Service.StreamEvents( + new StreamEventsRequest { SessionId = sessionId, AfterWorkerSequence = staleCursor }, + writer2, + new TestServerCallContext())); + + await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout); + + // sentinel + the retained tail (capacity events, all newer than the stale cursor). + int expected = 1 + capacity; + IReadOnlyList resumed = await writer2.WaitForMessageCountAsync(expected, TestTimeout); + + launcher.StopEmitting(); + await CloseAndDrainAsync(fixture, sessionId, stream2Task, launcher); + + // ---- the first message is the ReplayGap sentinel ---- + Assert.Equal(expected, resumed.Count); + + MxEvent sentinel = resumed[0]; + Assert.NotNull(sentinel.ReplayGap); + Assert.Equal(MxEventFamily.Unspecified, sentinel.Family); + Assert.Equal(MxEvent.BodyOneofCase.None, sentinel.BodyCase); + Assert.Equal(sessionId, sentinel.SessionId); + Assert.Equal(staleCursor, sentinel.ReplayGap.RequestedAfterSequence); + Assert.Equal(oldestAvailable, sentinel.ReplayGap.OldestAvailableSequence); + + // ---- the sentinel is followed by the retained tail, ascending, no further sentinels ---- + IReadOnlyList tail = resumed.Skip(1).ToArray(); + Assert.Equal(capacity, tail.Count); + Assert.All(tail, e => Assert.Null(e.ReplayGap)); + Assert.All(tail, e => Assert.True( + e.WorkerSequence >= oldestAvailable, + $"Retained event {e.WorkerSequence} is older than the oldest available {oldestAvailable}.")); + + ulong[] expectedTail = sequences.Skip(totalEvents - capacity).ToArray(); + Assert.Equal(expectedTail, tail.Select(e => e.WorkerSequence).ToArray()); + } + + // ---- shared flow helpers ---- + + private static async Task OpenSessionAsync( + ReconnectReplayGatewayServiceFixture fixture, + string name) + { + OpenSessionReply openReply = await fixture.Service.OpenSession( + new OpenSessionRequest + { + ClientSessionName = name, + ClientCorrelationId = $"open-{name}", + CommandTimeout = Duration.FromTimeSpan(TestTimeout), + }, + new TestServerCallContext()).ConfigureAwait(false); + + Assert.Equal(ProtocolStatusCode.Ok, openReply.ProtocolStatus.Code); + return openReply.SessionId; + } + + private static async Task WireUpAdviseAsync( + ReconnectReplayGatewayServiceFixture fixture, + string sessionId) + { + MxCommandReply registerReply = await fixture.Service.Invoke( + CreateRegisterRequest(sessionId), + new TestServerCallContext()).ConfigureAwait(false); + Assert.Equal(ProtocolStatusCode.Ok, registerReply.ProtocolStatus.Code); + + MxCommandReply addItemReply = await fixture.Service.Invoke( + CreateAddItemRequest(sessionId, registerReply.Register.ServerHandle), + new TestServerCallContext()).ConfigureAwait(false); + Assert.Equal(ProtocolStatusCode.Ok, addItemReply.ProtocolStatus.Code); + + MxCommandReply adviseReply = await fixture.Service.Invoke( + CreateAdviseRequest(sessionId, registerReply.Register.ServerHandle, addItemReply.AddItem.ItemHandle), + new TestServerCallContext()).ConfigureAwait(false); + Assert.Equal(ProtocolStatusCode.Ok, adviseReply.ProtocolStatus.Code); + } + + // Cancels the stream's token and awaits the stream task. Awaiting is load-bearing: it lets + // EventStreamService's finally block dispose the subscriber lease (dropping the session's + // active-subscriber count to zero) BEFORE the reconnect attaches, so the reconnect never + // races a still-registered subscriber. + private static async Task DetachAsync(CancellationTokenSource cts, Task streamTask) + { + await cts.CancelAsync().ConfigureAwait(false); + try + { + await streamTask.WaitAsync(TestTimeout).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // Expected: the iterator surfaces the cancellation. + } + catch (RpcException rpc) when (rpc.StatusCode == StatusCode.Cancelled) + { + // Also acceptable depending on gRPC exception wrapping. + } + } + + private static async Task CloseAndDrainAsync( + ReconnectReplayGatewayServiceFixture fixture, + string sessionId, + Task streamTask, + GatedEventFakeWorkerProcessLauncher launcher) + { + await fixture.Service.CloseSession( + new CloseSessionRequest { SessionId = sessionId, ClientCorrelationId = "close-reconnect" }, + new TestServerCallContext()).ConfigureAwait(false); + + try + { + await streamTask.WaitAsync(TestTimeout).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + } + + await launcher.WorkerTask.WaitAsync(TestTimeout).ConfigureAwait(false); + } + + // ---- request builders ---- + + private static MxCommandRequest CreateRegisterRequest(string sessionId) => + new() + { + SessionId = sessionId, + ClientCorrelationId = "register-rr", + Command = new MxCommand + { + Kind = MxCommandKind.Register, + Register = new RegisterCommand { ClientName = "reconnect-replay-e2e-client" }, + }, + }; + + private static MxCommandRequest CreateAddItemRequest(string sessionId, int serverHandle) => + new() + { + SessionId = sessionId, + ClientCorrelationId = "add-item-rr", + Command = new MxCommand + { + Kind = MxCommandKind.AddItem, + AddItem = new AddItemCommand + { + ServerHandle = serverHandle, + ItemDefinition = "Galaxy.Tag.Value", + }, + }, + }; + + private static MxCommandRequest CreateAdviseRequest( + string sessionId, + int serverHandle, + int itemHandle) => + new() + { + SessionId = sessionId, + ClientCorrelationId = "advise-rr", + Command = new MxCommand + { + Kind = MxCommandKind.Advise, + Advise = new AdviseCommand { ServerHandle = serverHandle, ItemHandle = itemHandle }, + }, + }; + + private static void ConfigureCommandReply(MxCommandReply reply, MxCommandKind kind) + { + switch (kind) + { + case MxCommandKind.Register: + reply.Register = new RegisterReply { ServerHandle = ServerHandle }; + break; + case MxCommandKind.AddItem: + reply.AddItem = new AddItemReply { ItemHandle = ItemHandle }; + break; + } + } + + // ---- fixture ---- + + /// + /// Gateway service fixture in single-subscriber mode with a configurable replay ring + /// capacity, so each test can retain the whole event history (no gap) or force capacity + /// eviction (gap). + /// + private sealed class ReconnectReplayGatewayServiceFixture : IAsyncDisposable + { + private readonly GatewayMetrics _metrics = new(); + private readonly SessionRegistry _registry = new(); + + /// Initializes a new instance of the class. + /// Fake worker process launcher backing the session manager. + /// Replay ring capacity for the session's event distributor. + public ReconnectReplayGatewayServiceFixture( + IWorkerProcessLauncher launcher, + int replayBufferCapacity) + { + IOptions options = Options.Create(CreateOptions(replayBufferCapacity)); + SessionWorkerClientFactory workerClientFactory = new( + launcher, + options, + _metrics, + NullLoggerFactory.Instance); + SessionManager sessionManager = new( + _registry, + workerClientFactory, + options, + _metrics, + logger: NullLogger.Instance, + dashboardEventBroadcaster: NullDashboardEventBroadcaster.Instance); + MxAccessGrpcMapper mapper = new(); + EventStreamService eventStreamService = new( + sessionManager, + options, + _metrics); + + Service = new MxAccessGatewayService( + sessionManager, + new GatewayRequestIdentityAccessor(), + new AllowAllConstraintEnforcer(), + new MxAccessGrpcRequestValidator(), + mapper, + eventStreamService, + _metrics, + NullLogger.Instance, + new FakeGatewayAlarmService()); + } + + /// Gets the gateway service under test. + public MxAccessGatewayService Service { get; } + + /// + /// Polls for + /// until it reaches , bounded by + /// . Fails the test on timeout. This is the deterministic + /// gate that proves the production code has (re)registered a subscriber before the + /// test drives events or reconnects. + /// + /// Identifier of the session to poll. + /// Target subscriber count to wait for. + /// Maximum time to wait before failing the test. + /// A task that represents the asynchronous operation. + public async Task WaitForSubscriberCountAsync(string sessionId, int n, TimeSpan timeout) + { + using CancellationTokenSource deadlineCts = new(timeout); + + while (true) + { + if (_registry.TryGet(sessionId, out GatewaySession? session) + && session.ActiveEventSubscriberCount >= n) + { + return; + } + + if (deadlineCts.IsCancellationRequested) + { + int actual = _registry.TryGet(sessionId, out GatewaySession? s) + ? s.ActiveEventSubscriberCount + : -1; + Assert.Fail( + $"Timed out waiting for {n} event subscriber(s) on session {sessionId}. " + + $"Actual count after {timeout.TotalSeconds:0.#}s: {actual}."); + } + + await Task.Delay(millisecondsDelay: 5, deadlineCts.Token).ConfigureAwait(false); + } + } + + /// Disposes every session in the registry and releases the fixture's metrics. + /// A task that represents the asynchronous operation. + public async ValueTask DisposeAsync() + { + foreach (GatewaySession session in _registry.Snapshot()) + { + await session.DisposeAsync().ConfigureAwait(false); + } + + _metrics.Dispose(); + } + + private static GatewayOptions CreateOptions(int replayBufferCapacity) => + new() + { + Worker = new WorkerOptions + { + StartupTimeoutSeconds = 5, + ShutdownTimeoutSeconds = 5, + HeartbeatIntervalSeconds = 30, + HeartbeatGraceSeconds = 30, + MaxMessageBytes = WorkerFrameProtocolOptions.DefaultMaxMessageBytes, + }, + Sessions = new SessionOptions + { + DefaultCommandTimeoutSeconds = 5, + MaxSessions = 4, + + // Single-subscriber mode: a fully-detached subscriber (stream task awaited) + // frees the sole slot, so the reconnect attaches cleanly without needing + // multi-subscriber fan-out. Detach-grace keeps the session Ready across the + // detach and the fixture runs no lease-reaper, so the session survives to be + // reconnected. + AllowMultipleEventSubscribers = false, + MaxEventSubscribersPerSession = 8, + }, + Events = new EventOptions + { + QueueCapacity = 32, + ReplayBufferCapacity = replayBufferCapacity, + + // Keep age-eviction effectively off for the duration of a fast test so + // capacity is the only eviction axis under test. + ReplayRetentionSeconds = 300, + }, + }; + } + + // ---- fake worker launcher ---- + + /// + /// Fake worker that emits events one at a time, gated by , so + /// the test drives event timing deterministically. Modeled on the multi-subscriber E2E + /// tests' gated launcher. The worker loop is independent of subscribers, so events emitted + /// while no gRPC stream is attached still flow through the distributor pump into the + /// session's replay ring — exactly the condition the reconnect/replay tests exercise. Call + /// before closing the session so the loop exits cleanly and can + /// process the shutdown envelope. + /// + private sealed class GatedEventFakeWorkerProcessLauncher : IWorkerProcessLauncher + { + public const int ProcessId = 7730; + + private readonly FakeWorkerProcess _process = new(ProcessId); + + // Capacity 64 so AllowNextEvent can be called ahead of time without blocking. + private readonly SemaphoreSlim _emitGate = new(0, 64); + private volatile bool _stopEmitting; + + /// Gets the task representing the fake worker's running background loop. + public Task WorkerTask { get; private set; } = Task.CompletedTask; + + /// Releases the gate so the worker emits one event. + public void AllowNextEvent() => _emitGate.Release(); + + /// + /// Signals the worker to stop waiting for the emit gate and process the shutdown + /// envelope. Must be called before CloseSession. + /// + public void StopEmitting() + { + _stopEmitting = true; + _emitGate.Release(); // unblock a pending gate wait if any + } + + /// + public Task LaunchAsync( + WorkerProcessLaunchRequest request, + CancellationToken cancellationToken = default) + { + WorkerTask = RunWorkerAsync(request, cancellationToken); + + return Task.FromResult(new WorkerProcessHandle( + _process, + new WorkerProcessCommandLine("reconnect-replay-fake-worker.exe", []), + DateTimeOffset.UtcNow)); + } + + private async Task RunWorkerAsync( + WorkerProcessLaunchRequest request, + CancellationToken cancellationToken) + { + await using FakeWorkerHarness harness = await FakeWorkerHarness.ConnectToGatewayPipeAsync( + request.SessionId, + request.Nonce, + request.PipeName, + request.ProtocolVersion, + cancellationToken: cancellationToken).ConfigureAwait(false); + await harness.CompleteStartupAsync(ProcessId, cancellationToken: cancellationToken).ConfigureAwait(false); + + int advisedServerHandle = 0; + int advisedItemHandle = 0; + int emittedCount = 0; + + while (!cancellationToken.IsCancellationRequested) + { + // While subscribed and not stopped, emit gated events using a non-blocking peek + // at the gate so incoming envelopes (including shutdown) are never starved. + while (advisedServerHandle != 0 + && !_stopEmitting + && await _emitGate.WaitAsync(millisecondsTimeout: 0).ConfigureAwait(false)) + { + int index = ++emittedCount; + await harness.EmitEventAsync( + MxEventFamily.OnDataChange, + cancellationToken, + mxEvent => + { + mxEvent.ServerHandle = advisedServerHandle; + mxEvent.ItemHandle = advisedItemHandle; + mxEvent.Quality = 192; + mxEvent.Value = new MxValue + { + DataType = MxDataType.String, + StringValue = $"reconnect-value-{index}", + }; + mxEvent.OnDataChange = new OnDataChangeEvent(); + }).ConfigureAwait(false); + } + + WorkerEnvelope? envelope; + try + { + using CancellationTokenSource readCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + readCts.CancelAfter(TimeSpan.FromMilliseconds(50)); + envelope = await harness.ReadGatewayEnvelopeAsync(readCts.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) + { + // Timed out waiting for an envelope — loop back to check the gate / emit. + continue; + } + + if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerShutdown) + { + await harness.SendShutdownAckAsync(cancellationToken: cancellationToken).ConfigureAwait(false); + _process.MarkExited(0); + return; + } + + if (envelope.BodyCase != WorkerEnvelope.BodyOneofCase.WorkerCommand) + { + throw new InvalidOperationException($"Unexpected envelope {envelope.BodyCase}."); + } + + MxCommand command = envelope.WorkerCommand.Command; + await harness.ReplyToCommandAsync( + envelope, + configureReply: reply => ConfigureCommandReply(reply, command.Kind), + cancellationToken: cancellationToken).ConfigureAwait(false); + + if (command.Kind == MxCommandKind.Advise) + { + advisedServerHandle = command.Advise.ServerHandle; + advisedItemHandle = command.Advise.ItemHandle; + } + } + } + } +}