1 changed files with 642 additions and 0 deletions
@@ -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;
 /// <summary>
 ///     End-to-end reconnect/replay tests through the real gRPC <c>StreamEvents</c> 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 <see cref="StreamEventsRequest.AfterWorkerSequence"/>. Covers both
 ///     the no-gap resume (cursor inside the retained window) and the <c>ReplayGap</c> sentinel
 ///     (cursor predates the oldest retained event after capacity eviction).
 /// </summary>
 /// <remarks>
 ///     These tests run in single-subscriber mode (<c>AllowMultipleEventSubscribers=false</c>).
 ///     Reconnect works because the first stream is FULLY detached before the reconnect attaches:
 ///     awaiting the first stream task runs <c>EventStreamService</c>'s finally block, which
 ///     disposes the subscriber lease and drops the session's active-subscriber count back to
 ///     zero, so the reconnect's <c>AttachEventSubscriberWithReplay</c> 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.
 /// </remarks>
 public sealed class GatewayEndToEndReconnectReplayTests
 {
    private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(10);
    private const int ServerHandle = 3001;
    private const int ItemHandle = 4002;
    /// <summary>
    ///     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 <c>ReplayGap</c>
    ///     sentinel.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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<MxEvent> 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<MxEvent> 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<MxEvent> 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<MxEvent> 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);
    }
    /// <summary>
    ///     Reconnecting with a cursor that predates the oldest retained event (after capacity
    ///     eviction) yields the <c>ReplayGap</c> sentinel FIRST — family unspecified, no body, no
    ///     per-item fields, correct requested/oldest sequences — followed by exactly the retained
    ///     tail, in ascending order.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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<MxEvent> 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<MxEvent> 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<MxEvent> 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<MxEvent> 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<MxEvent> 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<string> 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 ----
    /// <summary>
    ///     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).
    /// </summary>
    private sealed class ReconnectReplayGatewayServiceFixture : IAsyncDisposable
    {
        private readonly GatewayMetrics _metrics = new();
        private readonly SessionRegistry _registry = new();
        /// <summary>Initializes a new instance of the <see cref="ReconnectReplayGatewayServiceFixture"/> class.</summary>
        /// <param name="launcher">Fake worker process launcher backing the session manager.</param>
        /// <param name="replayBufferCapacity">Replay ring capacity for the session's event distributor.</param>
        public ReconnectReplayGatewayServiceFixture(
            IWorkerProcessLauncher launcher,
            int replayBufferCapacity)
        {
            IOptions<GatewayOptions> options = Options.Create(CreateOptions(replayBufferCapacity));
            SessionWorkerClientFactory workerClientFactory = new(
                launcher,
                options,
                _metrics,
                NullLoggerFactory.Instance);
            SessionManager sessionManager = new(
                _registry,
                workerClientFactory,
                options,
                _metrics,
                logger: NullLogger<SessionManager>.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<MxAccessGatewayService>.Instance,
                new FakeGatewayAlarmService());
        }
        /// <summary>Gets the gateway service under test.</summary>
        public MxAccessGatewayService Service { get; }
        /// <summary>
        ///     Polls <see cref="GatewaySession.ActiveEventSubscriberCount"/> for
        ///     <paramref name="sessionId"/> until it reaches <paramref name="n"/>, bounded by
        ///     <paramref name="timeout"/>. 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.
        /// </summary>
        /// <param name="sessionId">Identifier of the session to poll.</param>
        /// <param name="n">Target subscriber count to wait for.</param>
        /// <param name="timeout">Maximum time to wait before failing the test.</param>
        /// <returns>A task that represents the asynchronous operation.</returns>
        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);
            }
        }
        /// <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()
        {
            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 ----
    /// <summary>
    ///     Fake worker that emits events one at a time, gated by <see cref="AllowNextEvent"/>, 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
    ///     <see cref="StopEmitting"/> before closing the session so the loop exits cleanly and can
    ///     process the shutdown envelope.
    /// </summary>
    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;
        /// <summary>Gets the task representing the fake worker's running background loop.</summary>
        public Task WorkerTask { get; private set; } = Task.CompletedTask;
        /// <summary>Releases the gate so the worker emits one event.</summary>
        public void AllowNextEvent() => _emitGate.Release();
        /// <summary>
        ///     Signals the worker to stop waiting for the emit gate and process the shutdown
        ///     envelope. Must be called before <c>CloseSession</c>.
        /// </summary>
        public void StopEmitting()
        {
            _stopEmitting = true;
            _emitGate.Release(); // unblock a pending gate wait if any
        }
        /// <inheritdoc />
        public Task<WorkerProcessHandle> 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;
                }
            }
        }
    }
 }