perf(worker): launcher-configurable event queue capacity

src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs

@@ -10,6 +10,12 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase<GatewayOption
     private const int MinimumMaxMessageBytes = 1024;
     private const int MaximumMaxMessageBytes = 256 * 1024 * 1024;
 
+    // Bounds on the worker's outbound event-queue capacity. The floor keeps enough headroom that a
+    // normal subscription burst cannot overflow the queue (an overflow faults the whole session);
+    // the ceiling keeps a mistyped value from committing the x86 worker to an unbounded backlog.
+    private const int MinimumWorkerEventQueueCapacity = 1000;
+    private const int MaximumWorkerEventQueueCapacity = 1_000_000;
+
     // Whether the host is running in the Production environment. Drives the production-only
     // hard-stops (dashboard login disabled, plaintext LDAP transport) that must abort startup
     // rather than merely warn. Non-production hosts keep the permissive dev posture.
@@ -275,6 +281,12 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase<GatewayOption
                 "MxGateway:Worker:HeartbeatGraceSeconds must be greater than or equal to HeartbeatIntervalSeconds.");
         }
 
+        if (options.EventQueueCapacity is < MinimumWorkerEventQueueCapacity or > MaximumWorkerEventQueueCapacity)
+        {
+            builder.Add(
+                $"MxGateway:Worker:EventQueueCapacity must be between {MinimumWorkerEventQueueCapacity} and {MaximumWorkerEventQueueCapacity}.");
+        }
+
         if (options.MaxMessageBytes is < MinimumMaxMessageBytes or > MaximumMaxMessageBytes)
         {
             builder.Add(

src/ZB.MOM.WW.MxGateway.Server/Configuration/WorkerOptions.cs

@@ -33,6 +33,18 @@ public sealed class WorkerOptions
     /// </summary>
     public int WriteCompletionWaitMilliseconds { get; init; } = 1500;
 
+    /// <summary>
+    ///     Capacity of the worker's outbound MXAccess event queue, in events.
+    ///     Default 10,000; must be between 1,000 and 1,000,000. This is
+    ///     headroom, not a throttle: the queue has no drop policy, so a burst
+    ///     that fills it faults the session with a <c>QueueOverflow</c> worker
+    ///     fault. Raise it for sessions whose subscription set can outrun the
+    ///     drain loop (large advise sets, slow event consumers). Conveyed to
+    ///     the worker through the <c>MXGATEWAY_EVENT_QUEUE_CAPACITY</c>
+    ///     environment variable.
+    /// </summary>
+    public int EventQueueCapacity { get; init; } = 10000;
+
     /// <summary>The maximum time in seconds for graceful shutdown.</summary>
     public int ShutdownTimeoutSeconds { get; init; } = 10;
 

src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerProcessLauncher.cs

@@ -44,6 +44,13 @@ public sealed class WorkerProcessLauncher : IWorkerProcessLauncher
     public const string WorkerMaxAlarmsPerFetchEnvironmentVariableName =
         "MXGATEWAY_ALARM_MAX_ALARMS_PER_FETCH";
 
+    /// <summary>
+    /// Conveys MxGateway:Worker:EventQueueCapacity to the worker: the capacity
+    /// of the outbound MXAccess event queue, whose overflow faults the session.
+    /// </summary>
+    public const string WorkerEventQueueCapacityEnvironmentVariableName =
+        "MXGATEWAY_EVENT_QUEUE_CAPACITY";
+
     private readonly IWorkerProcessFactory _processFactory;
     private readonly IWorkerStartupProbe _startupProbe;
     private readonly GatewayMetrics _metrics;
@@ -202,6 +209,8 @@ public sealed class WorkerProcessLauncher : IWorkerProcessLauncher
             _workerOptions.PipeConnectAttemptTimeoutMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
         startInfo.Environment[WorkerWriteCompletionWaitEnvironmentVariableName] =
             _workerOptions.WriteCompletionWaitMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
+        startInfo.Environment[WorkerEventQueueCapacityEnvironmentVariableName] =
+            _workerOptions.EventQueueCapacity.ToString(System.Globalization.CultureInfo.InvariantCulture);
         startInfo.Environment[WorkerAlarmPollIntervalEnvironmentVariableName] =
             _alarmsOptions.PollIntervalMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
         startInfo.Environment[WorkerMaxAlarmsPerFetchEnvironmentVariableName] =

src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs

@@ -1099,4 +1099,59 @@ public sealed class GatewayOptionsValidatorTests
         Assert.True(result.Failed);
         Assert.Contains(result.Failures!, f => f.Contains("MaxMessageBytes") && f.Contains("reserve"));
     }
+
+    /// <summary>
+    /// Verifies the shipped worker event-queue capacity default passes validation and still matches
+    /// the worker-side <c>MxAccessEventQueue.DefaultCapacity</c> the environment variable falls back
+    /// to when the launcher value is unusable.
+    /// </summary>
+    [Fact]
+    public void Validate_Succeeds_WithDefaultEventQueueCapacity()
+    {
+        Assert.Equal(10000, new WorkerOptions().EventQueueCapacity);
+
+        ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, ValidOptions());
+        Assert.True(result.Succeeded);
+    }
+
+    /// <summary>
+    /// Verifies a worker event-queue capacity outside the supported range fails validation. Too
+    /// small leaves no burst headroom (an overflow faults the whole session); too large commits the
+    /// 32-bit worker to an outsized pre-allocation.
+    /// </summary>
+    /// <param name="eventQueueCapacity">Capacity under test.</param>
+    [Theory]
+    [InlineData(0)]
+    [InlineData(-1)]
+    [InlineData(999)]
+    [InlineData(1_000_001)]
+    public void Validate_Fails_WhenEventQueueCapacityOutOfRange(int eventQueueCapacity)
+    {
+        ValidateOptionsResult result = new GatewayOptionsValidator().Validate(
+            null,
+            WithWorkerAndProtocol(
+                new WorkerOptions { EventQueueCapacity = eventQueueCapacity },
+                new ProtocolOptions()));
+
+        Assert.True(result.Failed);
+        Assert.Contains(
+            result.Failures!,
+            f => f.Contains("MxGateway:Worker:EventQueueCapacity", StringComparison.Ordinal));
+    }
+
+    /// <summary>Verifies the worker event-queue capacity bounds themselves are accepted.</summary>
+    /// <param name="eventQueueCapacity">Capacity under test.</param>
+    [Theory]
+    [InlineData(1000)]
+    [InlineData(1_000_000)]
+    public void Validate_Succeeds_AtEventQueueCapacityBounds(int eventQueueCapacity)
+    {
+        ValidateOptionsResult result = new GatewayOptionsValidator().Validate(
+            null,
+            WithWorkerAndProtocol(
+                new WorkerOptions { EventQueueCapacity = eventQueueCapacity },
+                new ProtocolOptions()));
+
+        Assert.True(result.Succeeded);
+    }
 }

src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerProcessLauncherTests.cs

@@ -47,6 +47,13 @@ public sealed class WorkerProcessLauncherTests
             "1500",
             processFactory.LastStartInfo.Environment[
                 WorkerProcessLauncher.WorkerWriteCompletionWaitEnvironmentVariableName]);
+        // The worker sizes its outbound event queue from the launch environment;
+        // the queue has no drop policy, so this capacity is the session's burst
+        // headroom rather than a throttle.
+        Assert.Equal(
+            "10000",
+            processFactory.LastStartInfo.Environment[
+                WorkerProcessLauncher.WorkerEventQueueCapacityEnvironmentVariableName]);
         // MxGateway:Alarms defaults reach the worker's alarm poll loop and its
         // GetXmlCurrentAlarms2 cap (which is also its truncation threshold)
         // through the launch environment, not the command line.
@@ -64,6 +71,32 @@ public sealed class WorkerProcessLauncherTests
         Assert.Equal(0, metrics.GetSnapshot().WorkersRunning);
     }
 
+    /// <summary>
+    /// Verifies that a configured <see cref="WorkerOptions.EventQueueCapacity"/> — not the shipped
+    /// default — is what reaches the worker, so the option is deployable without a worker rebuild.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task LaunchAsync_WithConfiguredEventQueueCapacity_ExportsItToTheWorkerEnvironment()
+    {
+        using TestDirectory directory = TestDirectory.Create();
+        string executablePath = directory.CreateWorkerExecutable(machine: 0x014c);
+        FakeWorkerProcessFactory processFactory = new(new FakeWorkerProcess(processId: 1234));
+        WorkerProcessLauncher launcher = CreateLauncher(
+            executablePath,
+            processFactory,
+            new SucceedingStartupProbe(),
+            eventQueueCapacity: 65536);
+
+        using WorkerProcessHandle handle = await launcher.LaunchAsync(CreateRequest());
+
+        Assert.NotNull(processFactory.LastStartInfo);
+        Assert.Equal(
+            "65536",
+            processFactory.LastStartInfo.Environment[
+                WorkerProcessLauncher.WorkerEventQueueCapacityEnvironmentVariableName]);
+    }
+
     /// <summary>Verifies that a failed startup probe kills and disposes the worker process.</summary>
     /// <returns>A task that represents the asynchronous operation.</returns>
     [Fact]
@@ -216,7 +249,8 @@ public sealed class WorkerProcessLauncherTests
         GatewayMetrics? metrics = null,
         int startupTimeoutSeconds = 30,
         int startupProbeRetryAttempts = 3,
-        int startupProbeRetryDelayMilliseconds = 250)
+        int startupProbeRetryDelayMilliseconds = 250,
+        int eventQueueCapacity = 10000)
     {
         GatewayOptions options = new()
         {
@@ -227,6 +261,7 @@ public sealed class WorkerProcessLauncherTests
                 StartupTimeoutSeconds = startupTimeoutSeconds,
                 StartupProbeRetryAttempts = startupProbeRetryAttempts,
                 StartupProbeRetryDelayMilliseconds = startupProbeRetryDelayMilliseconds,
+                EventQueueCapacity = eventQueueCapacity,
             },
         };
 

src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs

@@ -419,6 +419,75 @@ public sealed class MxAccessEventQueueTests
         await Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await wait);
     }
 
+    /// <summary>
+    ///     The queue capacity comes from the launcher-set environment variable;
+    ///     a missing, unparseable, below-floor, or above-ceiling value must fall
+    ///     back to the 10,000 default rather than throw. A bad environment value
+    ///     must never stop the worker's session from starting.
+    /// </summary>
+    /// <param name="environmentValue">Raw environment value under test.</param>
+    /// <param name="expected">Expected resolved capacity.</param>
+    [Theory]
+    [InlineData(null, MxAccessEventQueue.DefaultCapacity)]
+    [InlineData("", MxAccessEventQueue.DefaultCapacity)]
+    [InlineData("not-a-number", MxAccessEventQueue.DefaultCapacity)]
+    [InlineData("0", MxAccessEventQueue.DefaultCapacity)]
+    [InlineData("-5", MxAccessEventQueue.DefaultCapacity)]
+    [InlineData("999", MxAccessEventQueue.DefaultCapacity)]
+    [InlineData("1000001", MxAccessEventQueue.DefaultCapacity)]
+    [InlineData("1000", 1000)]
+    [InlineData("1000000", 1000000)]
+    [InlineData("50000", 50000)]
+    public void ResolveCapacity_WithEnvironmentValue_FallsBackToDefaultWhenUnusable(
+        string? environmentValue,
+        int expected)
+    {
+        string? original = Environment.GetEnvironmentVariable(
+            MxAccessEventQueue.CapacityEnvironmentVariableName);
+        try
+        {
+            Environment.SetEnvironmentVariable(
+                MxAccessEventQueue.CapacityEnvironmentVariableName,
+                environmentValue);
+
+            Assert.Equal(expected, MxAccessEventQueue.ResolveCapacity());
+        }
+        finally
+        {
+            Environment.SetEnvironmentVariable(
+                MxAccessEventQueue.CapacityEnvironmentVariableName,
+                original);
+        }
+    }
+
+    /// <summary>
+    ///     A resolved capacity is what the queue is actually built with, so the
+    ///     configured headroom reaches the overflow check rather than only the
+    ///     resolver.
+    /// </summary>
+    [Fact]
+    public void Constructor_WithResolvedCapacity_UsesEnvironmentValue()
+    {
+        string? original = Environment.GetEnvironmentVariable(
+            MxAccessEventQueue.CapacityEnvironmentVariableName);
+        try
+        {
+            Environment.SetEnvironmentVariable(
+                MxAccessEventQueue.CapacityEnvironmentVariableName,
+                "2500");
+
+            MxAccessEventQueue queue = new(MxAccessEventQueue.ResolveCapacity());
+
+            Assert.Equal(2500, queue.Capacity);
+        }
+        finally
+        {
+            Environment.SetEnvironmentVariable(
+                MxAccessEventQueue.CapacityEnvironmentVariableName,
+                original);
+        }
+    }
+
     // Mirrors MxAccessEventQueue's per-event repeated-field allowance. Kept local rather than made
     // public on the queue: the byte-budget tests state their budgets in units of that charge, so a
     // change to it should surface here as a failing bound instead of silently moving with the code.

src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Globalization;
 using System.Threading;
 using System.Threading.Tasks;
 using Google.Protobuf.WellKnownTypes;
@@ -28,6 +29,31 @@ public sealed class MxAccessEventQueue
     /// </summary>
     public const int DefaultCapacity = 10000;
 
+    /// <summary>
+    ///     Environment variable the gateway's <c>WorkerProcessLauncher</c> sets
+    ///     from <c>MxGateway:Worker:EventQueueCapacity</c>. A missing,
+    ///     unparseable, or out-of-range value falls back to
+    ///     <see cref="DefaultCapacity"/> — a bad environment value must never
+    ///     keep the session from starting.
+    /// </summary>
+    internal const string CapacityEnvironmentVariableName = "MXGATEWAY_EVENT_QUEUE_CAPACITY";
+
+    /// <summary>
+    ///     Floor on the resolved capacity. Mirrors the gateway-side
+    ///     <c>MxGateway:Worker:EventQueueCapacity</c> minimum so a value that
+    ///     slipped past startup validation still leaves the headroom the queue
+    ///     needs: it has no drop policy, so an overflow faults the session.
+    /// </summary>
+    internal const int MinimumCapacity = 1000;
+
+    /// <summary>
+    ///     Ceiling on the resolved capacity, mirroring the gateway-side maximum.
+    ///     The backing queue pre-allocates its slots, so an absurd environment
+    ///     value would otherwise cost the 32-bit worker that allocation at
+    ///     startup.
+    /// </summary>
+    internal const int MaximumCapacity = 1_000_000;
+
     // Extra per-event slack added to WorkerEvent.CalculateSize() when charging the byte budget in
     // Drain(maxEvents, maxTotalBytes). CalculateSize() already accounts for the event's own tag and
     // length-delimiter (the WorkerEvent wrapper serializes MxEvent as field 1, and the reply packs
@@ -80,6 +106,28 @@ public sealed class MxAccessEventQueue
         events = new Queue<WorkerEvent>(capacity);
     }
 
+    /// <summary>
+    ///     Resolves the queue capacity from the launcher-provided environment
+    ///     variable. A missing, unparseable, or out-of-range value falls back to
+    ///     <see cref="DefaultCapacity"/> rather than throwing: running on the
+    ///     default capacity is always safe, and a worker that refuses to start
+    ///     over a mistyped environment variable is not.
+    /// </summary>
+    /// <returns>The capacity to construct the outbound event queue with.</returns>
+    internal static int ResolveCapacity()
+    {
+        string? value = Environment.GetEnvironmentVariable(CapacityEnvironmentVariableName);
+        return int.TryParse(
+                value,
+                NumberStyles.Integer,
+                CultureInfo.InvariantCulture,
+                out int capacity)
+            && capacity >= MinimumCapacity
+            && capacity <= MaximumCapacity
+                ? capacity
+                : DefaultCapacity;
+    }
+
     /// <summary>
     /// The queue's maximum capacity.
     /// </summary>

src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs

@@ -63,12 +63,15 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
 
     /// <summary>
     /// Initializes a new instance of <see cref="MxAccessStaSession"/> with default dependencies.
+    /// The outbound event queue is sized from the launcher-provided
+    /// <c>MXGATEWAY_EVENT_QUEUE_CAPACITY</c> (see <see cref="MxAccessEventQueue.ResolveCapacity"/>);
+    /// callers that pass their own queue keep full control of its capacity.
     /// </summary>
     public MxAccessStaSession()
         : this(
             new StaRuntime(),
             new MxAccessComObjectFactory(),
-            new MxAccessEventQueue())
+            new MxAccessEventQueue(MxAccessEventQueue.ResolveCapacity()))
     {
     }
 
@@ -84,7 +87,7 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
         : this(
             new StaRuntime(),
             new MxAccessComObjectFactory(),
-            new MxAccessEventQueue(),
+            new MxAccessEventQueue(MxAccessEventQueue.ResolveCapacity()),
             alarmCommandHandlerFactory)
     {
     }
@@ -99,7 +102,7 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
         StaRuntime staRuntime,
         IMxAccessComObjectFactory factory,
         IMxAccessEventSink eventSink)
-        : this(staRuntime, factory, eventSink, new MxAccessEventQueue())
+        : this(staRuntime, factory, eventSink, new MxAccessEventQueue(MxAccessEventQueue.ResolveCapacity()))
     {
     }