fix(dotnet-client): no ack retry, caller-owned channels, safe session dispose, typed diagnostics, AttachSession

2026-10-22 06:09:25 -04:00
parent 488649ce99
commit 6f0b408833
6 changed files with 495 additions and 27 deletions
@@ -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(