mxaccessgw/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionWorkerClientFactory.cs

using System.IO.Pipes;
using Microsoft.Extensions.Logging;
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.Metrics;
using ZB.MOM.WW.MxGateway.Server.Workers;

namespace ZB.MOM.WW.MxGateway.Server.Sessions;

/// <summary>Factory for creating worker clients and launching worker processes.</summary>
public sealed class SessionWorkerClientFactory : ISessionWorkerClientFactory
{
    /// <summary>
    /// Kernel buffer quota requested for each direction of a worker pipe. A zero quota — what the
    /// short <see cref="NamedPipeServerStream"/> overloads request — makes every byte-mode write
    /// rendezvous with a pending read, so writer latency is coupled to reader scheduling and a
    /// writer with no reader parked blocks indefinitely. That is the failure class behind the
    /// historical windev full-suite wedge (all tests reported, testhost never exiting). A real
    /// quota lets a whole frame land in the kernel and the writer return. 128 KiB comfortably
    /// holds the control traffic and typical event batches without reserving nonpaged pool per
    /// session for the rare maximum-sized frame, which still streams through in chunks.
    /// </summary>
    private const int PipeBufferSizeBytes = 128 * 1024;

    private readonly IWorkerProcessLauncher _workerProcessLauncher;
    private readonly GatewayMetrics _metrics;
    private readonly TimeProvider _timeProvider;
    private readonly ILoggerFactory _loggerFactory;
    private readonly GatewayOptions _options;

    /// <summary>Initializes a new instance of the SessionWorkerClientFactory class.</summary>
    /// <param name="workerProcessLauncher">Service for launching worker processes.</param>
    /// <param name="options">Configuration options.</param>
    /// <param name="metrics">Metrics collector for gateway events.</param>
    /// <param name="loggerFactory">Logger factory for creating loggers.</param>
    /// <param name="timeProvider">Optional time provider for testing; defaults to system time.</param>
    public SessionWorkerClientFactory(
        IWorkerProcessLauncher workerProcessLauncher,
        IOptions<GatewayOptions> options,
        GatewayMetrics metrics,
        ILoggerFactory loggerFactory,
        TimeProvider? timeProvider = null)
    {
        _workerProcessLauncher = workerProcessLauncher ?? throw new ArgumentNullException(nameof(workerProcessLauncher));
        ArgumentNullException.ThrowIfNull(options);
        _metrics = metrics ?? throw new ArgumentNullException(nameof(metrics));
        _loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory));
        _timeProvider = timeProvider ?? TimeProvider.System;
        _options = options.Value;
    }

    /// <inheritdoc />
    public async Task<IWorkerClient> CreateAsync(
        GatewaySession session,
        CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(session);

        NamedPipeServerStream? pipe = CreatePipe(session.PipeName);
        WorkerProcessHandle? processHandle = null;
        IWorkerClient? workerClient = null;
        using CancellationTokenSource startupCancellation =
            CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        startupCancellation.CancelAfter(session.StartupTimeout);
        try
        {
            session.TransitionTo(SessionState.StartingWorker);
            processHandle = await _workerProcessLauncher
                .LaunchAsync(
                    new WorkerProcessLaunchRequest(
                        session.SessionId,
                        session.PipeName,
                        GatewayContractInfo.WorkerProtocolVersion,
                        session.Nonce,
                        pipe),
                    startupCancellation.Token)
                .ConfigureAwait(false);

            session.TransitionTo(SessionState.WaitingForPipe);
            await WaitForPipeConnectionAsync(pipe, startupCancellation.Token).ConfigureAwait(false);

            session.TransitionTo(SessionState.Handshaking);
            WorkerFrameProtocolOptions frameOptions = new(
                session.SessionId,
                GatewayContractInfo.WorkerProtocolVersion,
                _options.Worker.MaxMessageBytes);
            WorkerClientConnection connection = new(
                session.SessionId,
                session.Nonce,
                pipe,
                frameOptions,
                processHandle);
            WorkerClientOptions clientOptions = new()
            {
                HeartbeatGrace = TimeSpan.FromSeconds(_options.Worker.HeartbeatGraceSeconds),
                HeartbeatCheckInterval = TimeSpan.FromSeconds(_options.Worker.HeartbeatIntervalSeconds),
                EventChannelCapacity = _options.Events.QueueCapacity,
                MaxPendingCommands = _options.Sessions.MaxPendingCommandsPerSession,
            };

            workerClient = new WorkerClient(
                connection,
                clientOptions,
                _metrics,
                _timeProvider,
                _loggerFactory.CreateLogger<WorkerClient>());

            pipe = null;
            processHandle = null;

            session.TransitionTo(SessionState.InitializingWorker);
            await workerClient.StartAsync(startupCancellation.Token).ConfigureAwait(false);

            return workerClient;
        }
        catch (Exception exception)
        {
            if (workerClient is not null)
            {
                try
                {
                    workerClient.Kill("OpenSessionFailed");
                }
                catch
                {
                    // Preserve the startup failure while still disposing below.
                }

                await workerClient.DisposeAsync().ConfigureAwait(false);
            }
            else
            {
                if (processHandle is not null)
                {
                    try
                    {
                        if (!processHandle.Process.HasExited)
                        {
                            processHandle.Process.Kill(entireProcessTree: true);
                            _metrics.WorkerKilled("OpenSessionFailed");
                        }
                    }
                    finally
                    {
                        processHandle.Dispose();
                    }
                }

                pipe?.Dispose();
            }

            if (exception is OperationCanceledException
                && startupCancellation.IsCancellationRequested
                && !cancellationToken.IsCancellationRequested)
            {
                throw new TimeoutException(
                    $"Worker session {session.SessionId} did not complete startup within {session.StartupTimeout}.",
                    exception);
            }

            throw;
        }
    }

    /// <summary>Creates a named pipe for worker communication.</summary>
    /// <param name="pipeName">The pipe name.</param>
    /// <returns>Named pipe server stream.</returns>
    /// <remarks>
    /// The buffer sizes are explicit so the pipe is not created with a zero quota; see
    /// <see cref="PipeBufferSizeBytes"/>. On Unix hosts (the macOS test matrix, where named pipes
    /// are Unix domain sockets) the sizes are advisory — the fix targets Windows production.
    /// </remarks>
    private static NamedPipeServerStream CreatePipe(string pipeName)
    {
        return new NamedPipeServerStream(
            pipeName,
            PipeDirection.InOut,
            maxNumberOfServerInstances: 1,
            PipeTransmissionMode.Byte,
            PipeOptions.Asynchronous,
            inBufferSize: PipeBufferSizeBytes,
            outBufferSize: PipeBufferSizeBytes);
    }

    /// <summary>Waits for a client to connect to the pipe.</summary>
    /// <param name="pipe">The named pipe.</param>
    /// <param name="cancellationToken">Cancellation token.</param>
    private static async Task WaitForPipeConnectionAsync(
        NamedPipeServerStream pipe,
        CancellationToken cancellationToken)
    {
        await pipe.WaitForConnectionAsync(cancellationToken).ConfigureAwait(false);
    }
}