perf(worker): launcher-configurable event queue capacity

docs/GatewayConfiguration.md

@@ -115,6 +115,7 @@ launch CWD (SEC-01, SEC-33).
 | `MxGateway:Worker:StartupProbeRetryDelayMilliseconds` | `250` | Delay between transient startup probe retry attempts. |
 | `MxGateway:Worker:PipeConnectAttemptTimeoutMilliseconds` | `2000` | Per-attempt timeout used by the worker named-pipe connect retry path. The overall pipe connection still stays under the startup budget. |
 | `MxGateway:Worker:WriteCompletionWaitMilliseconds` | `1500` | Bounded wait the worker holds a unary write reply (`Write`/`Write2`/`WriteSecured`/`WriteSecured2`; bulk writes excluded) for the matching MXAccess `OnWriteComplete` callback, so the reply's `statuses` carry the real commit outcome. `0` disables the wait (pure fire-and-forget replies). Must be `>= 0`. The gateway conveys the value to the worker via the `MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS` environment variable. Consumers that time their own writes must budget above this wait: OtOpcUa's GalaxyDriver wraps gateway writes in a 2 s Tier A resilience timeout, so a deployment raising this option past ~2000 must raise that driver `ResilienceConfig` write timeout in step or slow-but-successful commits surface as consumer-side failures. |
+| `MxGateway:Worker:EventQueueCapacity` | `10000` | Capacity, in events, of the worker's outbound MXAccess event queue. Must be between `1000` and `1000000`. This is burst headroom, not a throttle: the queue has no drop policy, so filling it records a `QueueOverflow` worker fault and faults the session. Raise it for sessions whose subscription set can outrun the drain loop (large advise sets, slow event consumers); the backing queue pre-allocates its slots, so the ceiling keeps a mistyped value from committing the 32-bit worker to an outsized allocation. The gateway conveys the value to the worker via the `MXGATEWAY_EVENT_QUEUE_CAPACITY` environment variable; a missing or unusable value leaves the worker on the 10000 default rather than failing the session. |
 | `MxGateway:Worker:ShutdownTimeoutSeconds` | `10` | Grace period for worker shutdown before the gateway treats shutdown as failed and may kill the worker process tree. |
 | `MxGateway:Worker:HeartbeatIntervalSeconds` | `5` | Worker heartbeat send interval and gateway heartbeat check cadence input. |
 | `MxGateway:Worker:HeartbeatGraceSeconds` | `15` | Maximum age of the last worker heartbeat before the gateway faults the worker. This must be greater than or equal to `HeartbeatIntervalSeconds`. |

docs/MxAccessWorkerInstanceDesign.md

@@ -416,7 +416,11 @@ type on buffered events. `OperationComplete` is only emitted from the native
 `MxAccessEventQueue` is the bounded outbound event queue for one worker
 session. It assigns the monotonic `WorkerSequence` and `WorkerTimestamp` when an
 event is accepted, preserving the order in which MXAccess handlers enqueue
-events. The default capacity is `10000`. When the queue reaches capacity it
+events. The capacity is `10000` by default and comes from
+`MxGateway:Worker:EventQueueCapacity`, which the gateway stamps onto the worker
+launch environment as `MXGATEWAY_EVENT_QUEUE_CAPACITY`; a missing, unparseable,
+or out-of-range value (outside `1000`–`1000000`) leaves the worker on the
+default rather than failing the session. When the queue reaches capacity it
 records a `WorkerFaultCategory.QueueOverflow` fault and rejects further events.
 The event handler catches conversion and enqueue failures, records the first
 fault on the queue, and returns to the STA message pump instead of writing to

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()))
     {
     }