From 6f0b408833d5653272c060f5f8a6fcd8de6d8fec Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:09:25 -0400 Subject: [PATCH] fix(dotnet-client): no ack retry, caller-owned channels, safe session dispose, typed diagnostics, AttachSession --- clients/dotnet/README.md | 24 ++ .../MxGatewayClientAlarmsTests.cs | 26 +- .../MxGatewayClientRemediationTests.cs | 241 ++++++++++++++++++ .../GalaxyRepositoryClient.cs | 38 ++- .../MxGatewayClient.cs | 76 +++++- .../MxGatewaySession.cs | 117 ++++++++- 6 files changed, 495 insertions(+), 27 deletions(-) create mode 100644 clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientRemediationTests.cs diff --git a/clients/dotnet/README.md b/clients/dotnet/README.md index d651ddf..415329d 100644 --- a/clients/dotnet/README.md +++ b/clients/dotnet/README.md @@ -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` into the same `MxEventStreamItem` surface. diff --git a/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientAlarmsTests.cs b/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientAlarmsTests.cs index 803dbec..9a002b6 100644 --- a/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientAlarmsTests.cs +++ b/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientAlarmsTests.cs @@ -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(() => - 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); } /// AcknowledgeAlarmAsync maps unauthenticated RPC exception to typed exception. diff --git a/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientRemediationTests.cs b/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientRemediationTests.cs new file mode 100644 index 0000000..c317f67 --- /dev/null +++ b/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientRemediationTests.cs @@ -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; + +/// Tests for ack no-retry, caller-owned channels, safe dispose, typed diagnostics, and AttachSession. +public sealed class MxGatewayClientRemediationTests +{ + /// An acknowledge is a non-idempotent operator write and is never retried. + /// A task that represents the asynchronous operation. + [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(() => client.AcknowledgeAlarmAsync( + new AcknowledgeAlarmRequest { OperatorUser = "op", OperatorPassword = "pw" })); + + Assert.Single(transport.AcknowledgeAlarmCalls); + } + + /// A client created over a caller-owned channel does not dispose it. + /// A task that represents the asynchronous operation. + [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( + 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(ex); + } + + /// Create over an existing channel rejects a null channel. + [Fact] + public void Create_WithNullChannel_Throws() + { + Assert.Throws(() => MxGatewayClient.Create(null!, Options())); + Assert.Throws(() => GalaxyRepositoryClient.Create(null!, Options())); + } + + /// DisposeAsync swallows a gateway failure during close and still disposes. + /// A task that represents the asynchronous operation. + [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(() => session.CloseAsync()); + } + + /// DisposeAsync after the client is disposed does not throw. + /// A task that represents the asynchronous operation. + [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(() => session.CloseAsync()); + } + + /// PingAsync sends a Ping command carrying the message. + /// A task that represents the asynchronous operation. + [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); + } + + /// GetSessionStateAsync returns the typed state. + /// A task that represents the asynchronous operation. + [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); + } + + /// GetSessionStateAsync throws on a reply without payload. + /// A task that represents the asynchronous operation. + [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(() => session.GetSessionStateAsync()); + } + + /// GetWorkerInfoAsync returns the worker info payload. + /// A task that represents the asynchronous operation. + [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); + } + + /// DrainEventsAsync sends max_events and returns the events. + /// A task that represents the asynchronous operation. + [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 events = await session.DrainEventsAsync(10); + + Assert.Equal(5UL, Assert.Single(events).WorkerSequence); + Assert.Equal(10U, Assert.Single(transport.InvokeCalls).Request.Command.DrainEvents.MaxEvents); + } + + /// DrainEventsAsync treats a missing payload as an empty drain. + /// A task that represents the asynchronous operation. + [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()); + } + + /// AttachSession binds to an existing id without opening a session and resumes the stream. + /// A task that represents the asynchronous operation. + [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 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); + } + + /// AttachSession rejects blank ids and a disposed client. + /// A task that represents the asynchronous operation. + [Fact] + public async Task AttachSession_InvalidArguments_Throw() + { + FakeGatewayTransport transport = CreateTransport(); + MxGatewayClient client = new(transport.Options, transport); + + Assert.Throws(() => client.AttachSession(" ")); + await client.DisposeAsync(); + Assert.Throws(() => 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()); +} diff --git a/clients/dotnet/ZB.MOM.WW.MxGateway.Client/GalaxyRepositoryClient.cs b/clients/dotnet/ZB.MOM.WW.MxGateway.Client/GalaxyRepositoryClient.cs index ee19b21..a259fe3 100644 --- a/clients/dotnet/ZB.MOM.WW.MxGateway.Client/GalaxyRepositoryClient.cs +++ b/clients/dotnet/ZB.MOM.WW.MxGateway.Client/GalaxyRepositoryClient.cs @@ -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); + } + + /// + /// Creates a Galaxy Repository client over a caller-owned . + /// The client does not dispose the channel; the caller controls its lifetime and any + /// TLS/handler configuration. + /// + /// An existing channel, shared or pre-configured by the caller. + /// Client options (API key, timeouts, retry). + /// A new client instance that does not own the channel. + 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); } /// @@ -436,7 +464,11 @@ public sealed class GalaxyRepositoryClient : IAsyncDisposable } _disposed = true; - _channel?.Dispose(); + if (_ownsChannel) + { + _channel?.Dispose(); + } + return ValueTask.CompletedTask; } diff --git a/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewayClient.cs b/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewayClient.cs index 9ed1dcd..cb33125 100644 --- a/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewayClient.cs +++ b/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewayClient.cs @@ -14,7 +14,8 @@ namespace ZB.MOM.WW.MxGateway.Client; /// 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()); - _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); + } + + /// + /// Creates a gateway client over a caller-owned . The client + /// does not dispose the channel; the caller controls its lifetime and any TLS/handler + /// configuration. is still validated but + /// the channel's own address is used for transport. + /// + /// An existing channel, shared or pre-configured by the caller. + /// Client configuration options (API key, timeouts, retry). + /// A new gateway client instance that does not own the channel. + 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); + } + + /// + /// Binds a session object to an existing gateway session id without calling + /// OpenSession, so a reconnecting process can resume + /// with afterWorkerSequence. + /// 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 carries only the + /// session id. + /// + /// The id of a session previously opened with this client's API key. + /// A session bound to . + public MxGatewaySession AttachSession(string sessionId) + { + ArgumentException.ThrowIfNullOrWhiteSpace(sessionId); + ThrowIfDisposed(); + + return new MxGatewaySession( + this, + new OpenSessionReply + { + SessionId = sessionId, + ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok }, + }); } /// @@ -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; } diff --git a/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewaySession.cs b/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewaySession.cs index 8a9e570..1aae534 100644 --- a/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewaySession.cs +++ b/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewaySession.cs @@ -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); } + /// + /// Sends a PING diagnostic command to the worker and throws if the reply is a failure. + /// + /// Optional message echoed to the worker; defaults to empty. + /// Cancellation token. + /// A task that completes when the worker acknowledges the ping. + 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(); + } + + /// + /// Gets the gateway's lifecycle state for this session. + /// + /// Cancellation token. + /// The current session state. + public async Task 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"); + } + + /// + /// Gets diagnostic information about the worker process serving this session. + /// + /// Cancellation token. + /// The worker process id, version, and MXAccess COM identifiers. + public async Task 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"); + } + + /// + /// Drains events queued in the worker through the diagnostic command path. This removes + /// events from the worker queue; it is not a substitute for + /// and is not retried automatically. The reply is bounded by server-side caps, so call + /// repeatedly until it returns an empty list. + /// + /// Maximum events to return; 0 lets the server choose. + /// Cancellation token. + /// The drained events, possibly empty. + public async Task> 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()); + } + /// /// Streams events from the worker for this session, optionally starting after a given sequence number. /// @@ -1505,8 +1606,20 @@ public sealed class MxGatewaySession : IAsyncDisposable /// A task that represents the asynchronous operation. 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 InvokeCommandAsync(