Merge stream F (.NET client: no ack retry, shared channel, safe dispose, typed diagnostics, AttachSession)

2026-10-22 06:09:55 -04:00
parent c18566bcf2 6f0b408833
commit 3a1ee28729
6 changed files with 495 additions and 27 deletions
@@ -137,6 +137,30 @@ gateway's own sentinel observable. Normal events pass through with
 `IsReplayGap == false` and `ReplayGap == null`. The gap is only ever produced by
 `StreamEvents`; the diagnostic drain path never emits it. If you already consume
 the raw `StreamEventsAsync` (or the client-level stream), the
+### Caller-owned channels, resuming sessions, diagnostics
+
+`MxGatewayClient.Create(GrpcChannel, MxGatewayClientOptions)` and
+`GalaxyRepositoryClient.Create(GrpcChannel, MxGatewayClientOptions)` build a
+client over a channel the caller created (shared channel, custom handler or
+TLS). The client never disposes a caller-owned channel; `Create(options)` still
+creates and owns its own channel.
+
+`MxGatewayClient.AttachSession(sessionId)` returns an `MxGatewaySession` bound
+to an existing session id without calling OpenSession, so a restarted process
+can resume `StreamEventsAsync(afterWorkerSequence)`. The gateway only allows the
+API key that opened the session to attach; any other key gets `PermissionDenied`.
+The attached session's `OpenSessionReply` carries only the session id.
+
+`MxGatewaySession` has typed diagnostics: `PingAsync`, `GetSessionStateAsync`,
+`GetWorkerInfoAsync`, and `DrainEventsAsync` (removes queued events from the
+worker; call until empty). `DisposeAsync` closes the session best-effort and does
+not throw when the gateway is unreachable or the client is already disposed.
+
+Retry applies only to idempotent calls (`CloseSessionRawAsync`, ping,
+session-state and worker-info commands, Galaxy reads). `AcknowledgeAlarmAsync`
+is never retried because it is a non-idempotent operator write that re-sends the
+operator password and can double-acknowledge in the Historian.
+
 `AsStreamItemsAsync()` extension projects any `IAsyncEnumerable<MxEvent>` into
 the same `MxEventStreamItem` surface.

@@ -81,24 +81,22 @@ public sealed class MxGatewayClientAlarmsTests
    [Fact]
    public async Task AcknowledgeAlarmAsync_HonorsCancellation()
    {
-        // Acks are routed through the safe-unary retry pipeline (idempotent at the
-        // MxAccess level), so the transport-side cancellation token is a linked one
-        // rather than the caller's original. Verify cancellation by tripping the source
-        // and asserting the call observes it.
+        // Acks bypass the retry pipeline (non-idempotent), so the caller's token
+        // reaches the transport unchanged.
        using CancellationTokenSource cancellation = new();
-        cancellation.Cancel();
        FakeGatewayTransport transport = CreateTransport();
        await using MxGatewayClient client = CreateClient(transport);

-        await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
-            client.AcknowledgeAlarmAsync(
-                new AcknowledgeAlarmRequest
-                {
-                    AlarmFullReference = "Tank01.Level.HiHi",
-                    Comment = string.Empty,
-                    OperatorUser = "alice",
-                },
-                cancellation.Token));
+        await client.AcknowledgeAlarmAsync(
+            new AcknowledgeAlarmRequest
+            {
+                AlarmFullReference = "Tank01.Level.HiHi",
+                Comment = string.Empty,
+                OperatorUser = "alice",
+            },
+            cancellation.Token);
+
+        Assert.Equal(cancellation.Token, Assert.Single(transport.AcknowledgeAlarmCalls).CallOptions.CancellationToken);
    }

    /// <summary>AcknowledgeAlarmAsync maps unauthenticated RPC exception to typed exception.</summary>
@@ -0,0 +1,241 @@
+using Grpc.Core;
+using Grpc.Net.Client;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+
+namespace ZB.MOM.WW.MxGateway.Client.Tests;
+
+/// <summary>Tests for ack no-retry, caller-owned channels, safe dispose, typed diagnostics, and AttachSession.</summary>
+public sealed class MxGatewayClientRemediationTests
+{
+    /// <summary>An acknowledge is a non-idempotent operator write and is never retried.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task AcknowledgeAlarmAsync_DoesNotRetryTransientFailure()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        transport.AcknowledgeAlarmExceptions.Enqueue(new RpcException(new Status(StatusCode.Unavailable, "down")));
+        await using MxGatewayClient client = new(transport.Options, transport);
+
+        await Assert.ThrowsAsync<RpcException>(() => client.AcknowledgeAlarmAsync(
+            new AcknowledgeAlarmRequest { OperatorUser = "op", OperatorPassword = "pw" }));
+
+        Assert.Single(transport.AcknowledgeAlarmCalls);
+    }
+
+    /// <summary>A client created over a caller-owned channel does not dispose it.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Create_WithExistingChannel_DoesNotDisposeChannel()
+    {
+        using GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:1");
+        MxGatewayClient gateway = MxGatewayClient.Create(channel, Options());
+        GalaxyRepositoryClient galaxy = GalaxyRepositoryClient.Create(channel, Options());
+
+        await gateway.DisposeAsync();
+        await galaxy.DisposeAsync();
+
+        // A disposed channel throws ObjectDisposedException when a call is created.
+        var invoker = channel.CreateCallInvoker();
+        Exception? ex = Record.Exception(() => invoker.AsyncUnaryCall(
+            new Method<byte[], byte[]>(
+                MethodType.Unary,
+                "s",
+                "m",
+                Marshallers.Create(b => b, b => b),
+                Marshallers.Create(b => b, b => b)),
+            null,
+            new CallOptions(deadline: DateTime.UtcNow.AddMilliseconds(1)),
+            []).Dispose());
+        Assert.IsNotType<ObjectDisposedException>(ex);
+    }
+
+    /// <summary>Create over an existing channel rejects a null channel.</summary>
+    [Fact]
+    public void Create_WithNullChannel_Throws()
+    {
+        Assert.Throws<ArgumentNullException>(() => MxGatewayClient.Create(null!, Options()));
+        Assert.Throws<ArgumentNullException>(() => GalaxyRepositoryClient.Create(null!, Options()));
+    }
+
+    /// <summary>DisposeAsync swallows a gateway failure during close and still disposes.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Session_DisposeAsync_SwallowsCloseFailure()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        transport.CloseSessionExceptions.Enqueue(new MxGatewayException("unreachable"));
+        await using MxGatewayClient client = new(transport.Options, transport);
+        MxGatewaySession session = await client.OpenSessionAsync();
+
+        await session.DisposeAsync();
+
+        Assert.Single(transport.CloseSessionCalls);
+        await Assert.ThrowsAsync<ObjectDisposedException>(() => session.CloseAsync());
+    }
+
+    /// <summary>DisposeAsync after the client is disposed does not throw.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Session_DisposeAsync_AfterClientDisposed_DoesNotThrow()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        MxGatewayClient client = new(transport.Options, transport);
+        MxGatewaySession session = await client.OpenSessionAsync();
+        await client.DisposeAsync();
+
+        await session.DisposeAsync();
+
+        await Assert.ThrowsAsync<ObjectDisposedException>(() => session.CloseAsync());
+    }
+
+    /// <summary>PingAsync sends a Ping command carrying the message.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task PingAsync_SendsPingCommand()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        transport.AddInvokeReply(OkReply(MxCommandKind.Ping));
+        await using MxGatewayClient client = new(transport.Options, transport);
+        MxGatewaySession session = await client.OpenSessionAsync();
+
+        await session.PingAsync("hello");
+
+        MxCommandRequest request = Assert.Single(transport.InvokeCalls).Request;
+        Assert.Equal(MxCommandKind.Ping, request.Command.Kind);
+        Assert.Equal("hello", request.Command.Ping.Message);
+        Assert.Equal("session-fixture", request.SessionId);
+    }
+
+    /// <summary>GetSessionStateAsync returns the typed state.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task GetSessionStateAsync_ReturnsState()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        MxCommandReply reply = OkReply(MxCommandKind.GetSessionState);
+        reply.SessionState = new SessionStateReply { State = SessionState.Ready };
+        transport.AddInvokeReply(reply);
+        await using MxGatewayClient client = new(transport.Options, transport);
+        MxGatewaySession session = await client.OpenSessionAsync();
+
+        SessionState state = await session.GetSessionStateAsync();
+
+        Assert.Equal(SessionState.Ready, state);
+        Assert.Equal(MxCommandKind.GetSessionState, Assert.Single(transport.InvokeCalls).Request.Command.Kind);
+    }
+
+    /// <summary>GetSessionStateAsync throws on a reply without payload.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task GetSessionStateAsync_MissingPayload_ThrowsMalformed()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        transport.AddInvokeReply(OkReply(MxCommandKind.GetSessionState));
+        await using MxGatewayClient client = new(transport.Options, transport);
+        MxGatewaySession session = await client.OpenSessionAsync();
+
+        await Assert.ThrowsAsync<MxGatewayMalformedReplyException>(() => session.GetSessionStateAsync());
+    }
+
+    /// <summary>GetWorkerInfoAsync returns the worker info payload.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task GetWorkerInfoAsync_ReturnsWorkerInfo()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        MxCommandReply reply = OkReply(MxCommandKind.GetWorkerInfo);
+        reply.WorkerInfo = new WorkerInfoReply { WorkerProcessId = 42, WorkerVersion = "1.2.3" };
+        transport.AddInvokeReply(reply);
+        await using MxGatewayClient client = new(transport.Options, transport);
+        MxGatewaySession session = await client.OpenSessionAsync();
+
+        WorkerInfoReply info = await session.GetWorkerInfoAsync();
+
+        Assert.Equal(42, info.WorkerProcessId);
+        Assert.Equal("1.2.3", info.WorkerVersion);
+    }
+
+    /// <summary>DrainEventsAsync sends max_events and returns the events.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task DrainEventsAsync_ReturnsEventsAndSendsMax()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        MxCommandReply reply = OkReply(MxCommandKind.DrainEvents);
+        reply.DrainEvents = new DrainEventsReply();
+        reply.DrainEvents.Events.Add(new MxEvent { SessionId = "session-fixture", WorkerSequence = 5 });
+        transport.AddInvokeReply(reply);
+        await using MxGatewayClient client = new(transport.Options, transport);
+        MxGatewaySession session = await client.OpenSessionAsync();
+
+        IReadOnlyList<MxEvent> events = await session.DrainEventsAsync(10);
+
+        Assert.Equal(5UL, Assert.Single(events).WorkerSequence);
+        Assert.Equal(10U, Assert.Single(transport.InvokeCalls).Request.Command.DrainEvents.MaxEvents);
+    }
+
+    /// <summary>DrainEventsAsync treats a missing payload as an empty drain.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task DrainEventsAsync_MissingPayload_ReturnsEmpty()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        transport.AddInvokeReply(OkReply(MxCommandKind.DrainEvents));
+        await using MxGatewayClient client = new(transport.Options, transport);
+        MxGatewaySession session = await client.OpenSessionAsync();
+
+        Assert.Empty(await session.DrainEventsAsync());
+    }
+
+    /// <summary>AttachSession binds to an existing id without opening a session and resumes the stream.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task AttachSession_BindsExistingIdWithoutOpening_AndResumesStream()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        transport.AddEvent(new MxEvent { SessionId = "existing", WorkerSequence = 8 });
+        await using MxGatewayClient client = new(transport.Options, transport);
+
+        MxGatewaySession session = client.AttachSession("existing");
+        List<MxEvent> received = [];
+        await foreach (MxEvent e in session.StreamEventsAsync(afterWorkerSequence: 7))
+        {
+            received.Add(e);
+        }
+
+        Assert.Empty(transport.OpenSessionCalls);
+        Assert.Equal("existing", session.SessionId);
+        StreamEventsRequest request = Assert.Single(transport.StreamEventsCalls).Request;
+        Assert.Equal("existing", request.SessionId);
+        Assert.Equal(7UL, request.AfterWorkerSequence);
+        Assert.Single(received);
+    }
+
+    /// <summary>AttachSession rejects blank ids and a disposed client.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task AttachSession_InvalidArguments_Throw()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        MxGatewayClient client = new(transport.Options, transport);
+
+        Assert.Throws<ArgumentException>(() => client.AttachSession(" "));
+        await client.DisposeAsync();
+        Assert.Throws<ObjectDisposedException>(() => client.AttachSession("x"));
+    }
+
+    private static MxCommandReply OkReply(MxCommandKind kind) => new()
+    {
+        SessionId = "session-fixture",
+        Kind = kind,
+        ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
+    };
+
+    private static MxGatewayClientOptions Options() => new()
+    {
+        Endpoint = new Uri("http://localhost:5000"),
+        ApiKey = "test-api-key",
+    };
+
+    private static FakeGatewayTransport CreateTransport() => new(Options());
+}
@@ -22,6 +22,7 @@ public sealed class GalaxyRepositoryClient : IAsyncDisposable
    private const int BrowseChildrenPageSize = 500;

    private readonly GrpcChannel? _channel;
+    private readonly bool _ownsChannel;
    private readonly IGalaxyRepositoryClientTransport _transport;
    private readonly ResiliencePipeline _safeUnaryRetryPipeline;
    private bool _disposed;
@@ -48,9 +49,11 @@ public sealed class GalaxyRepositoryClient : IAsyncDisposable

    private GalaxyRepositoryClient(
        GrpcChannel channel,
-        IGalaxyRepositoryClientTransport transport)
+        IGalaxyRepositoryClientTransport transport,
+        bool ownsChannel)
    {
        _channel = channel;
+        _ownsChannel = ownsChannel;
        _transport = transport;
        Options = transport.Options;
        _safeUnaryRetryPipeline = MxGatewayClientRetryPolicy.Create(
@@ -95,7 +98,32 @@ public sealed class GalaxyRepositoryClient : IAsyncDisposable
            channel,
            new GrpcGalaxyRepositoryClientTransport(
                options,
-                new GalaxyRepository.GalaxyRepositoryClient(channel)));
+                new GalaxyRepository.GalaxyRepositoryClient(channel)),
+            ownsChannel: true);
+    }
+
+    /// <summary>
+    /// Creates a Galaxy Repository client over a caller-owned <see cref="GrpcChannel"/>.
+    /// The client does not dispose the channel; the caller controls its lifetime and any
+    /// TLS/handler configuration.
+    /// </summary>
+    /// <param name="channel">An existing channel, shared or pre-configured by the caller.</param>
+    /// <param name="options">Client options (API key, timeouts, retry).</param>
+    /// <returns>A new client instance that does not own the channel.</returns>
+    public static GalaxyRepositoryClient Create(
+        GrpcChannel channel,
+        MxGatewayClientOptions options)
+    {
+        ArgumentNullException.ThrowIfNull(channel);
+        ArgumentNullException.ThrowIfNull(options);
+        options.Validate();
+
+        return new GalaxyRepositoryClient(
+            channel,
+            new GrpcGalaxyRepositoryClientTransport(
+                options,
+                new GalaxyRepository.GalaxyRepositoryClient(channel)),
+            ownsChannel: false);
    }

    /// <summary>
@@ -436,7 +464,11 @@ public sealed class GalaxyRepositoryClient : IAsyncDisposable
        }

        _disposed = true;
-        _channel?.Dispose();
+        if (_ownsChannel)
+        {
+            _channel?.Dispose();
+        }
+
        return ValueTask.CompletedTask;
    }

@@ -14,7 +14,8 @@ namespace ZB.MOM.WW.MxGateway.Client;
 /// </summary>
 public sealed class MxGatewayClient : IAsyncDisposable
 {
-    private readonly GrpcChannel _channel;
+    private readonly GrpcChannel? _channel;
+    private readonly bool _ownsChannel;
    private readonly IMxGatewayClientTransport _transport;
    private readonly ResiliencePipeline _safeUnaryRetryPipeline;
    private bool _disposed;
@@ -36,14 +37,16 @@ public sealed class MxGatewayClient : IAsyncDisposable
        _safeUnaryRetryPipeline = MxGatewayClientRetryPolicy.Create(
            options.Retry,
            options.LoggerFactory?.CreateLogger<MxGatewayClient>());
-        _channel = null!;
+        _channel = null;
    }

    private MxGatewayClient(
        GrpcChannel channel,
-        IMxGatewayClientTransport transport)
+        IMxGatewayClientTransport transport,
+        bool ownsChannel)
    {
        _channel = channel;
+        _ownsChannel = ownsChannel;
        _transport = transport;
        Options = transport.Options;
        _safeUnaryRetryPipeline = MxGatewayClientRetryPolicy.Create(
@@ -88,7 +91,58 @@ public sealed class MxGatewayClient : IAsyncDisposable
            channel,
            new GrpcMxGatewayClientTransport(
                options,
-                new MxAccessGateway.MxAccessGatewayClient(channel)));
+                new MxAccessGateway.MxAccessGatewayClient(channel)),
+            ownsChannel: true);
+    }
+
+    /// <summary>
+    /// Creates a gateway client over a caller-owned <see cref="GrpcChannel"/>. The client
+    /// does not dispose the channel; the caller controls its lifetime and any TLS/handler
+    /// configuration. <see cref="MxGatewayClientOptions.Endpoint"/> is still validated but
+    /// the channel's own address is used for transport.
+    /// </summary>
+    /// <param name="channel">An existing channel, shared or pre-configured by the caller.</param>
+    /// <param name="options">Client configuration options (API key, timeouts, retry).</param>
+    /// <returns>A new gateway client instance that does not own the channel.</returns>
+    public static MxGatewayClient Create(
+        GrpcChannel channel,
+        MxGatewayClientOptions options)
+    {
+        ArgumentNullException.ThrowIfNull(channel);
+        ArgumentNullException.ThrowIfNull(options);
+        options.Validate();
+
+        return new MxGatewayClient(
+            channel,
+            new GrpcMxGatewayClientTransport(
+                options,
+                new MxAccessGateway.MxAccessGatewayClient(channel)),
+            ownsChannel: false);
+    }
+
+    /// <summary>
+    /// Binds a session object to an existing gateway session id without calling
+    /// OpenSession, so a reconnecting process can resume
+    /// <see cref="MxGatewaySession.StreamEventsAsync"/> with <c>afterWorkerSequence</c>.
+    /// The gateway only allows the API key that opened the session to stream its events
+    /// or invoke commands on it; any other key is rejected with PermissionDenied. The
+    /// returned session's <see cref="MxGatewaySession.OpenSessionReply"/> carries only the
+    /// session id.
+    /// </summary>
+    /// <param name="sessionId">The id of a session previously opened with this client's API key.</param>
+    /// <returns>A session bound to <paramref name="sessionId"/>.</returns>
+    public MxGatewaySession AttachSession(string sessionId)
+    {
+        ArgumentException.ThrowIfNullOrWhiteSpace(sessionId);
+        ThrowIfDisposed();
+
+        return new MxGatewaySession(
+            this,
+            new OpenSessionReply
+            {
+                SessionId = sessionId,
+                ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
+            });
    }

    /// <summary>
@@ -199,11 +253,13 @@ public sealed class MxGatewayClient : IAsyncDisposable
        ArgumentNullException.ThrowIfNull(request);
        ThrowIfDisposed();

+        // Deliberately not wrapped in the retry pipeline: an acknowledge is a
+        // non-idempotent operator write that re-sends the operator password and
+        // can double-ack in the Historian.
        try
        {
-            return await ExecuteSafeUnaryAsync(
-                token => _transport.AcknowledgeAlarmAsync(request, CreateCallOptions(token)),
-                cancellationToken).ConfigureAwait(false);
+            return await _transport.AcknowledgeAlarmAsync(request, CreateCallOptions(cancellationToken))
+                .ConfigureAwait(false);
        }
        catch (MxGatewayException ex) when (request.OperatorPassword.Length > 0)
        {
@@ -266,7 +322,11 @@ public sealed class MxGatewayClient : IAsyncDisposable
        }

        _disposed = true;
-        _channel?.Dispose();
+        if (_ownsChannel)
+        {
+            _channel?.Dispose();
+        }
+
        return ValueTask.CompletedTask;
    }

@@ -1,3 +1,4 @@
+using Grpc.Core;
 using ZB.MOM.WW.MxGateway.Contracts.Proto;

 namespace ZB.MOM.WW.MxGateway.Client;
@@ -1452,6 +1453,106 @@ public sealed class MxGatewaySession : IAsyncDisposable
        return _client.InvokeAsync(request, cancellationToken);
    }

+    /// <summary>
+    /// Sends a PING diagnostic command to the worker and throws if the reply is a failure.
+    /// </summary>
+    /// <param name="message">Optional message echoed to the worker; defaults to empty.</param>
+    /// <param name="cancellationToken">Cancellation token.</param>
+    /// <returns>A task that completes when the worker acknowledges the ping.</returns>
+    public async Task PingAsync(
+        string? message = null,
+        CancellationToken cancellationToken = default)
+    {
+        MxCommandReply reply = await InvokeCommandAsync(
+                new MxCommand
+                {
+                    Kind = MxCommandKind.Ping,
+                    Ping = new PingCommand { Message = message ?? string.Empty },
+                },
+                cancellationToken)
+            .ConfigureAwait(false);
+        reply.EnsureProtocolSuccess().EnsureMxAccessSuccess();
+    }
+
+    /// <summary>
+    /// Gets the gateway's lifecycle state for this session.
+    /// </summary>
+    /// <param name="cancellationToken">Cancellation token.</param>
+    /// <returns>The current session state.</returns>
+    public async Task<SessionState> GetSessionStateAsync(CancellationToken cancellationToken = default)
+    {
+        MxCommandReply reply = await InvokeCommandAsync(
+                new MxCommand
+                {
+                    Kind = MxCommandKind.GetSessionState,
+                    GetSessionState = new GetSessionStateCommand(),
+                },
+                cancellationToken)
+            .ConfigureAwait(false);
+        reply.EnsureProtocolSuccess().EnsureMxAccessSuccess();
+
+        return reply.SessionState?.State
+            ?? throw CreateMalformed(reply, "GetSessionState");
+    }
+
+    /// <summary>
+    /// Gets diagnostic information about the worker process serving this session.
+    /// </summary>
+    /// <param name="cancellationToken">Cancellation token.</param>
+    /// <returns>The worker process id, version, and MXAccess COM identifiers.</returns>
+    public async Task<WorkerInfoReply> GetWorkerInfoAsync(CancellationToken cancellationToken = default)
+    {
+        MxCommandReply reply = await InvokeCommandAsync(
+                new MxCommand
+                {
+                    Kind = MxCommandKind.GetWorkerInfo,
+                    GetWorkerInfo = new GetWorkerInfoCommand(),
+                },
+                cancellationToken)
+            .ConfigureAwait(false);
+        reply.EnsureProtocolSuccess().EnsureMxAccessSuccess();
+
+        return reply.WorkerInfo
+            ?? throw CreateMalformed(reply, "GetWorkerInfo");
+    }
+
+    /// <summary>
+    /// Drains events queued in the worker through the diagnostic command path. This removes
+    /// events from the worker queue; it is not a substitute for <see cref="StreamEventsAsync"/>
+    /// and is not retried automatically. The reply is bounded by server-side caps, so call
+    /// repeatedly until it returns an empty list.
+    /// </summary>
+    /// <param name="maxEvents">Maximum events to return; 0 lets the server choose.</param>
+    /// <param name="cancellationToken">Cancellation token.</param>
+    /// <returns>The drained events, possibly empty.</returns>
+    public async Task<IReadOnlyList<MxEvent>> DrainEventsAsync(
+        uint maxEvents = 0,
+        CancellationToken cancellationToken = default)
+    {
+        MxCommandReply reply = await InvokeCommandAsync(
+                new MxCommand
+                {
+                    Kind = MxCommandKind.DrainEvents,
+                    DrainEvents = new DrainEventsCommand { MaxEvents = maxEvents },
+                },
+                cancellationToken)
+            .ConfigureAwait(false);
+        reply.EnsureProtocolSuccess().EnsureMxAccessSuccess();
+
+        return reply.DrainEvents?.Events.ToArray() ?? [];
+    }
+
+    private static MxGatewayMalformedReplyException CreateMalformed(MxCommandReply reply, string operation)
+    {
+        return new MxGatewayMalformedReplyException(
+            $"{operation} returned a malformed reply: OK reply carried no typed payload",
+            reply.SessionId,
+            reply.CorrelationId,
+            reply.ProtocolStatus,
+            reply.HasHresult ? reply.Hresult : null,
+            reply.Statuses.ToArray());
+    }
+
    /// <summary>
    /// Streams events from the worker for this session, optionally starting after a given sequence number.
    /// </summary>
@@ -1505,8 +1606,20 @@ public sealed class MxGatewaySession : IAsyncDisposable
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async ValueTask DisposeAsync()
    {
-        await CloseAsync().ConfigureAwait(false);
-        _closeLock.Dispose();
+        try
+        {
+            await CloseAsync().ConfigureAwait(false);
+        }
+        catch (Exception ex) when (ex is MxGatewayException or ObjectDisposedException or RpcException)
+        {
+            // Best-effort close: the gateway may be unreachable or the owning client
+            // already disposed. Disposal must not throw or leak the lock; the gateway
+            // reaps the session on its own idle/detach timers.
+        }
+        finally
+        {
+            _closeLock.Dispose();
+        }
    }

    private Task<MxCommandReply> InvokeCommandAsync(