fix(IPC-36,IPC-37): guard secondary contracts

This commit is contained in:
Joseph Doherty
2026-08-25 11:14:14 -04:00
parent f3827c71c9
commit d6c5a11ae6
19 changed files with 256 additions and 66 deletions
+16
View File
@@ -94,6 +94,15 @@ only and does not share types with `mxaccess_gateway.proto`. See
[Galaxy Repository Browse](./GalaxyRepository.md) for the RPC catalog and [Galaxy Repository Browse](./GalaxyRepository.md) for the RPC catalog and
behavior. behavior.
The retained proto remains the client-generation source of truth while the
gateway serves the wire-identical descriptor from the
`ZB.MOM.WW.GalaxyRepository` package. `GalaxyContractParityTests` converts both
loaded descriptors to `FileDescriptorProto`, removes source locations, and
normalizes the C# namespace and file name before comparing deterministic bytes.
Its mutation test also requires drift failures to identify the first differing
protobuf message. Any parity failure must be resolved before publishing either
Contracts or clients.
Generated C# output is written to `src/ZB.MOM.WW.MxGateway.Contracts/Generated/`. Do not Generated C# output is written to `src/ZB.MOM.WW.MxGateway.Contracts/Generated/`. Do not
hand-edit generated files. hand-edit generated files.
@@ -140,6 +149,13 @@ Run the focused contract tests after changing either `.proto` file:
dotnet test src/ZB.MOM.WW.MxGateway.Tests/ZB.MOM.WW.MxGateway.Tests.csproj --filter ProtobufContractRoundTripTests dotnet test src/ZB.MOM.WW.MxGateway.Tests/ZB.MOM.WW.MxGateway.Tests.csproj --filter ProtobufContractRoundTripTests
``` ```
When changing `galaxy_repository.proto` or the served Galaxy Repository
package, also run the retained-vs-served parity guard:
```bash
dotnet test src/ZB.MOM.WW.MxGateway.Tests/ZB.MOM.WW.MxGateway.Tests.csproj --filter GalaxyContractParityTests
```
The full solution build also regenerates the C# contracts before compiling The full solution build also regenerates the C# contracts before compiling
gateway and test projects: gateway and test projects:
+13 -3
View File
@@ -90,13 +90,22 @@ Expected command-line arguments:
--protocol-version <version> --protocol-version <version>
``` ```
Expected protected environment values: Expected launch environment values (the shared names are defined by
`WorkerLaunchEnvironment` in Contracts):
```text ```text
MXGATEWAY_WORKER_NONCE=<random nonce> MXGATEWAY_WORKER_NONCE=<random nonce>
MXGATEWAY_WORKER_LOG_CONTEXT=<optional context> MXGATEWAY_WORKER_PIPE_CONNECT_ATTEMPT_TIMEOUT_MS=<milliseconds>
MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS=<milliseconds>
MXGATEWAY_EVENT_QUEUE_CAPACITY=<events>
MXGATEWAY_ALARM_POLL_INTERVAL_MS=<milliseconds>
MXGATEWAY_ALARM_MAX_ALARMS_PER_FETCH=<alarms>
``` ```
The nonce is protected bootstrap material. The remaining values are derived
from the public `MxGateway:Worker:*` and `MxGateway:Alarms:*` configuration
keys documented in [Worker Process Launcher](./WorkerProcessLauncher.md).
Startup sequence: Startup sequence:
1. Parse command-line arguments. 1. Parse command-line arguments.
@@ -312,7 +321,8 @@ write wait still yields the empty-`statuses` unconfirmed reply, and an expired
per-tag `ReadBulk` wait still reports its own timeout. per-tag `ReadBulk` wait still reports its own timeout.
The write wait's budget is `MxGateway:Worker:WriteCompletionWaitMilliseconds` The write wait's budget is `MxGateway:Worker:WriteCompletionWaitMilliseconds`
(default 1500). It is a bounded hold on the STA per unary write, so deployments (default 1500). It applies to `Write`, `Write2`, `WriteSecured`, and
`WriteSecured2` and is a bounded hold on the STA per single-item write, so deployments
whose write workload is effectively fire-and-forget — no consumer reads the whose write workload is effectively fire-and-forget — no consumer reads the
reply's `statuses` — can lower it, or set `0` to skip the wait entirely and reply's `statuses` — can lower it, or set `0` to skip the wait entirely and
reply on acceptance alone. reply on acceptance alone.
+19 -5
View File
@@ -26,6 +26,23 @@ The launcher sets the nonce through the `MXGATEWAY_WORKER_NONCE` environment
variable. The nonce is not included in `WorkerProcessCommandLine` so logs and variable. The nonce is not included in `WorkerProcessCommandLine` so logs and
diagnostics can report the launch command without exposing the secret. diagnostics can report the launch command without exposing the secret.
All launch-variable names live in the shared Contracts
`WorkerLaunchEnvironment` class and are consumed by both the launcher and the
worker. The launcher always exports all six values:
| Environment variable | Gateway source | Worker default when missing/invalid |
|---|---|---|
| `MXGATEWAY_WORKER_NONCE` | `WorkerProcessLaunchRequest.Nonce` | None; bootstrap fails because the nonce is required |
| `MXGATEWAY_WORKER_PIPE_CONNECT_ATTEMPT_TIMEOUT_MS` | `MxGateway:Worker:PipeConnectAttemptTimeoutMilliseconds` | `2000` ms |
| `MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS` | `MxGateway:Worker:WriteCompletionWaitMilliseconds` | `1500` ms |
| `MXGATEWAY_EVENT_QUEUE_CAPACITY` | `MxGateway:Worker:EventQueueCapacity` | `10000` events |
| `MXGATEWAY_ALARM_POLL_INTERVAL_MS` | `MxGateway:Alarms:PollIntervalMilliseconds` | `500` ms |
| `MXGATEWAY_ALARM_MAX_ALARMS_PER_FETCH` | `MxGateway:Alarms:MaxAlarmsPerFetch` | `1024` alarms |
The write-completion wait applies to the four single-item write variants:
`Write`, `Write2`, `WriteSecured`, and `WriteSecured2`. Bulk writes report COM
acceptance per entry and do not wait for `OnWriteComplete` rows.
## Validation And Cleanup ## Validation And Cleanup
Before starting the process, the launcher validates that the configured worker Before starting the process, the launcher validates that the configured worker
@@ -44,11 +61,8 @@ process remains alive. The policy is configured by
`WorkerOptions.StartupProbeRetryDelayMilliseconds`; the retry counter is `WorkerOptions.StartupProbeRetryDelayMilliseconds`; the retry counter is
recorded as `mxgateway.retries.attempted` with `area=worker_startup`. recorded as `mxgateway.retries.attempted` with `area=worker_startup`.
The launcher also passes The worker uses the exported pipe-attempt value inside its own bounded Polly
`MXGATEWAY_WORKER_PIPE_CONNECT_ATTEMPT_TIMEOUT_MS` to the worker process from retry loop.
`WorkerOptions.PipeConnectAttemptTimeoutMilliseconds`. The worker uses that
value as the per-attempt named-pipe connect timeout inside its own bounded
Polly retry loop.
If startup fails or exceeds `WorkerOptions.StartupTimeoutSeconds`, the launcher If startup fails or exceeds `WorkerOptions.StartupTimeoutSeconds`, the launcher
kills the worker process tree, disposes the process handle, disposes the kills the worker process tree, disposes the process handle, disposes the
@@ -0,0 +1,25 @@
namespace ZB.MOM.WW.MxGateway.Contracts;
/// <summary>Names of environment variables shared by the gateway worker launcher and worker process.</summary>
public static class WorkerLaunchEnvironment
{
/// <summary>Bootstrap nonce used to authenticate the worker handshake.</summary>
public const string WorkerNonce = "MXGATEWAY_WORKER_NONCE";
/// <summary>Per-attempt named-pipe connection timeout in milliseconds.</summary>
public const string PipeConnectAttemptTimeoutMilliseconds =
"MXGATEWAY_WORKER_PIPE_CONNECT_ATTEMPT_TIMEOUT_MS";
/// <summary>Bounded wait for a Write/Write2/WriteSecured/WriteSecured2 completion callback in milliseconds.</summary>
public const string WriteCompletionWaitMilliseconds =
"MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS";
/// <summary>Capacity of the worker's outbound MXAccess event queue.</summary>
public const string EventQueueCapacity = "MXGATEWAY_EVENT_QUEUE_CAPACITY";
/// <summary>Worker alarm polling cadence in milliseconds.</summary>
public const string AlarmPollIntervalMilliseconds = "MXGATEWAY_ALARM_POLL_INTERVAL_MS";
/// <summary>Maximum alarm records requested in one current-alarm fetch.</summary>
public const string MaxAlarmsPerFetch = "MXGATEWAY_ALARM_MAX_ALARMS_PER_FETCH";
}
@@ -37,7 +37,9 @@
the wire-identical Galaxy types from the ZB.MOM.WW.GalaxyRepository package. This proto is the wire-identical Galaxy types from the ZB.MOM.WW.GalaxyRepository package. This proto is
still the generation source of truth for the language clients (Go/Rust/Java/Python compile it still the generation source of truth for the language clients (Go/Rust/Java/Python compile it
by path; the .NET client consumes Contracts.Proto.Galaxy.* via this project reference) and for by path; the .NET client consumes Contracts.Proto.Galaxy.* via this project reference) and for
clients/proto/proto-inputs.json. Removing it breaks all five clients — do not delete. --> clients/proto/proto-inputs.json. GalaxyContractParityTests normalizes its descriptor against
the descriptor served by the package and fails on wire drift. Removing it breaks all five
clients — do not delete. -->
<Protobuf Include="Protos\galaxy_repository.proto" ProtoRoot="Protos" OutputDir="Generated" GrpcOutputDir="Generated" GrpcServices="Both" /> <Protobuf Include="Protos\galaxy_repository.proto" ProtoRoot="Protos" OutputDir="Generated" GrpcOutputDir="Generated" GrpcServices="Both" />
</ItemGroup> </ItemGroup>
@@ -2,6 +2,7 @@ using System.Diagnostics;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Server.Configuration; using ZB.MOM.WW.MxGateway.Server.Configuration;
using ZB.MOM.WW.MxGateway.Server.Metrics; using ZB.MOM.WW.MxGateway.Server.Metrics;
using Polly; using Polly;
@@ -15,41 +16,47 @@ namespace ZB.MOM.WW.MxGateway.Server.Workers;
public sealed class WorkerProcessLauncher : IWorkerProcessLauncher public sealed class WorkerProcessLauncher : IWorkerProcessLauncher
{ {
/// <summary>Environment variable for worker nonce.</summary> /// <summary>Environment variable for worker nonce.</summary>
public const string WorkerNonceEnvironmentVariableName = "MXGATEWAY_WORKER_NONCE"; [Obsolete("Use WorkerLaunchEnvironment.WorkerNonce.")]
public const string WorkerNonceEnvironmentVariableName = WorkerLaunchEnvironment.WorkerNonce;
/// <summary>Environment variable for worker pipe connect attempt timeout.</summary> /// <summary>Environment variable for worker pipe connect attempt timeout.</summary>
[Obsolete("Use WorkerLaunchEnvironment.PipeConnectAttemptTimeoutMilliseconds.")]
public const string WorkerPipeConnectAttemptTimeoutEnvironmentVariableName = public const string WorkerPipeConnectAttemptTimeoutEnvironmentVariableName =
"MXGATEWAY_WORKER_PIPE_CONNECT_ATTEMPT_TIMEOUT_MS"; WorkerLaunchEnvironment.PipeConnectAttemptTimeoutMilliseconds;
/// <summary> /// <summary>
/// Conveys MxGateway:Worker:WriteCompletionWaitMilliseconds to the worker: /// Conveys MxGateway:Worker:WriteCompletionWaitMilliseconds to the worker:
/// the bounded wait for the OnWriteComplete callback on /// the bounded wait for the OnWriteComplete callback on Write, Write2,
/// WriteSecured/WriteSecured2 replies. 0 disables the wait. /// WriteSecured, and WriteSecured2 replies. 0 disables the wait.
/// </summary> /// </summary>
[Obsolete("Use WorkerLaunchEnvironment.WriteCompletionWaitMilliseconds.")]
public const string WorkerWriteCompletionWaitEnvironmentVariableName = public const string WorkerWriteCompletionWaitEnvironmentVariableName =
"MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS"; WorkerLaunchEnvironment.WriteCompletionWaitMilliseconds;
/// <summary> /// <summary>
/// Conveys MxGateway:Alarms:PollIntervalMilliseconds to the worker: the /// Conveys MxGateway:Alarms:PollIntervalMilliseconds to the worker: the
/// cadence at which the worker's STA polls the AVEVA alarm consumer. /// cadence at which the worker's STA polls the AVEVA alarm consumer.
/// </summary> /// </summary>
[Obsolete("Use WorkerLaunchEnvironment.AlarmPollIntervalMilliseconds.")]
public const string WorkerAlarmPollIntervalEnvironmentVariableName = public const string WorkerAlarmPollIntervalEnvironmentVariableName =
"MXGATEWAY_ALARM_POLL_INTERVAL_MS"; WorkerLaunchEnvironment.AlarmPollIntervalMilliseconds;
/// <summary> /// <summary>
/// Conveys MxGateway:Alarms:MaxAlarmsPerFetch to the worker: the cap /// Conveys MxGateway:Alarms:MaxAlarmsPerFetch to the worker: the cap
/// passed to GetXmlCurrentAlarms2, which is also the record count at /// passed to GetXmlCurrentAlarms2, which is also the record count at
/// which the worker treats a fetch as truncated. /// which the worker treats a fetch as truncated.
/// </summary> /// </summary>
[Obsolete("Use WorkerLaunchEnvironment.MaxAlarmsPerFetch.")]
public const string WorkerMaxAlarmsPerFetchEnvironmentVariableName = public const string WorkerMaxAlarmsPerFetchEnvironmentVariableName =
"MXGATEWAY_ALARM_MAX_ALARMS_PER_FETCH"; WorkerLaunchEnvironment.MaxAlarmsPerFetch;
/// <summary> /// <summary>
/// Conveys MxGateway:Worker:EventQueueCapacity to the worker: the capacity /// Conveys MxGateway:Worker:EventQueueCapacity to the worker: the capacity
/// of the outbound MXAccess event queue, whose overflow faults the session. /// of the outbound MXAccess event queue, whose overflow faults the session.
/// </summary> /// </summary>
[Obsolete("Use WorkerLaunchEnvironment.EventQueueCapacity.")]
public const string WorkerEventQueueCapacityEnvironmentVariableName = public const string WorkerEventQueueCapacityEnvironmentVariableName =
"MXGATEWAY_EVENT_QUEUE_CAPACITY"; WorkerLaunchEnvironment.EventQueueCapacity;
private readonly IWorkerProcessFactory _processFactory; private readonly IWorkerProcessFactory _processFactory;
private readonly IWorkerStartupProbe _startupProbe; private readonly IWorkerStartupProbe _startupProbe;
@@ -204,16 +211,16 @@ public sealed class WorkerProcessLauncher : IWorkerProcessLauncher
startInfo.ArgumentList.Add(argument); startInfo.ArgumentList.Add(argument);
} }
startInfo.Environment[WorkerNonceEnvironmentVariableName] = request.Nonce; startInfo.Environment[WorkerLaunchEnvironment.WorkerNonce] = request.Nonce;
startInfo.Environment[WorkerPipeConnectAttemptTimeoutEnvironmentVariableName] = startInfo.Environment[WorkerLaunchEnvironment.PipeConnectAttemptTimeoutMilliseconds] =
_workerOptions.PipeConnectAttemptTimeoutMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture); _workerOptions.PipeConnectAttemptTimeoutMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
startInfo.Environment[WorkerWriteCompletionWaitEnvironmentVariableName] = startInfo.Environment[WorkerLaunchEnvironment.WriteCompletionWaitMilliseconds] =
_workerOptions.WriteCompletionWaitMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture); _workerOptions.WriteCompletionWaitMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
startInfo.Environment[WorkerEventQueueCapacityEnvironmentVariableName] = startInfo.Environment[WorkerLaunchEnvironment.EventQueueCapacity] =
_workerOptions.EventQueueCapacity.ToString(System.Globalization.CultureInfo.InvariantCulture); _workerOptions.EventQueueCapacity.ToString(System.Globalization.CultureInfo.InvariantCulture);
startInfo.Environment[WorkerAlarmPollIntervalEnvironmentVariableName] = startInfo.Environment[WorkerLaunchEnvironment.AlarmPollIntervalMilliseconds] =
_alarmsOptions.PollIntervalMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture); _alarmsOptions.PollIntervalMilliseconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
startInfo.Environment[WorkerMaxAlarmsPerFetchEnvironmentVariableName] = startInfo.Environment[WorkerLaunchEnvironment.MaxAlarmsPerFetch] =
_alarmsOptions.MaxAlarmsPerFetch.ToString(System.Globalization.CultureInfo.InvariantCulture); _alarmsOptions.MaxAlarmsPerFetch.ToString(System.Globalization.CultureInfo.InvariantCulture);
commandLine = new WorkerProcessCommandLine(executablePath, arguments); commandLine = new WorkerProcessCommandLine(executablePath, arguments);
@@ -0,0 +1,100 @@
using Google.Protobuf;
using Google.Protobuf.Reflection;
using RetainedGalaxyReflection = ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy.GalaxyRepositoryReflection;
using ServedGalaxyReflection = ZB.MOM.WW.GalaxyRepository.Grpc.GalaxyRepositoryReflection;
namespace ZB.MOM.WW.MxGateway.Tests.Contracts;
public sealed class GalaxyContractParityTests
{
/// <summary>Guards the retained client proto against drift from the descriptor served by the gateway.</summary>
[Fact]
public void RetainedClientDescriptor_MatchesServedGatewayDescriptor()
{
FileDescriptorProto retained = Normalize(RetainedGalaxyReflection.Descriptor);
FileDescriptorProto served = Normalize(ServedGalaxyReflection.Descriptor);
AssertEquivalentDescriptors(retained, served);
}
/// <summary>Proves descriptor drift diagnostics name the first differing protobuf message.</summary>
[Fact]
public void DescriptorComparison_WhenMessageDiffers_ReportsDifferingMessage()
{
FileDescriptorProto retained = Normalize(RetainedGalaxyReflection.Descriptor);
FileDescriptorProto changed = retained.Clone();
DescriptorProto changedMessage = changed.MessageType.First(message => message.Field.Count > 0);
changedMessage.Field[0].Name += "_changed";
Xunit.Sdk.XunitException exception = Assert.Throws<Xunit.Sdk.XunitException>(
() => AssertEquivalentDescriptors(retained, changed));
Assert.Contains(changedMessage.Name, exception.Message, StringComparison.Ordinal);
}
private static FileDescriptorProto Normalize(FileDescriptor descriptor)
{
FileDescriptorProto proto = descriptor.ToProto();
proto.SourceCodeInfo = null;
proto.Name = string.Empty;
if (proto.Options is not null)
{
proto.Options.CsharpNamespace = string.Empty;
}
return proto;
}
private static void AssertEquivalentDescriptors(FileDescriptorProto expected, FileDescriptorProto actual)
{
byte[] expectedBytes = SerializeDeterministically(expected);
byte[] actualBytes = SerializeDeterministically(actual);
if (expectedBytes.AsSpan().SequenceEqual(actualBytes))
{
return;
}
throw new Xunit.Sdk.XunitException(
$"Galaxy Repository descriptors differ at {DescribeFirstDifference(expected, actual)}.");
}
private static string DescribeFirstDifference(FileDescriptorProto expected, FileDescriptorProto actual)
{
int comparableMessageCount = Math.Min(expected.MessageType.Count, actual.MessageType.Count);
for (int index = 0; index < comparableMessageCount; index++)
{
DescriptorProto expectedMessage = expected.MessageType[index];
DescriptorProto actualMessage = actual.MessageType[index];
if (!SerializeDeterministically(expectedMessage).AsSpan()
.SequenceEqual(SerializeDeterministically(actualMessage)))
{
string messageName = string.IsNullOrEmpty(expectedMessage.Name)
? actualMessage.Name
: expectedMessage.Name;
return $"message '{Qualify(expected.Package, messageName)}'";
}
}
if (expected.MessageType.Count != actual.MessageType.Count)
{
return "the top-level message list";
}
return "the normalized file descriptor";
}
private static string Qualify(string package, string symbol) =>
string.IsNullOrEmpty(package) ? symbol : package + "." + symbol;
private static byte[] SerializeDeterministically(IMessage message)
{
using MemoryStream stream = new();
using CodedOutputStream output = new(stream, leaveOpen: true)
{
Deterministic = true,
};
message.WriteTo(output);
output.Flush();
return stream.ToArray();
}
}
@@ -14,6 +14,16 @@ public sealed class WorkerProcessLauncherTests
private const string PipeName = "mxaccess-gateway-123-session-1"; private const string PipeName = "mxaccess-gateway-123-session-1";
private const string Nonce = "super-secret-nonce"; private const string Nonce = "super-secret-nonce";
private static readonly string[] RequiredLaunchEnvironmentVariables =
[
WorkerLaunchEnvironment.WorkerNonce,
WorkerLaunchEnvironment.PipeConnectAttemptTimeoutMilliseconds,
WorkerLaunchEnvironment.WriteCompletionWaitMilliseconds,
WorkerLaunchEnvironment.EventQueueCapacity,
WorkerLaunchEnvironment.AlarmPollIntervalMilliseconds,
WorkerLaunchEnvironment.MaxAlarmsPerFetch,
];
/// <summary>Verifies that a valid worker executable starts with correct bootstrap arguments and nonce environment variable.</summary> /// <summary>Verifies that a valid worker executable starts with correct bootstrap arguments and nonce environment variable.</summary>
/// <returns>A task that represents the asynchronous operation.</returns> /// <returns>A task that represents the asynchronous operation.</returns>
[Fact] [Fact]
@@ -38,33 +48,38 @@ public sealed class WorkerProcessLauncherTests
Assert.Equal( Assert.Equal(
["--session-id", SessionId, "--pipe-name", PipeName, "--protocol-version", "1"], ["--session-id", SessionId, "--pipe-name", PipeName, "--protocol-version", "1"],
processFactory.LastStartInfo.ArgumentList); processFactory.LastStartInfo.ArgumentList);
Assert.Equal(Nonce, processFactory.LastStartInfo.Environment[WorkerProcessLauncher.WorkerNonceEnvironmentVariableName]); Assert.Equal(Nonce, processFactory.LastStartInfo.Environment[WorkerLaunchEnvironment.WorkerNonce]);
Assert.Equal( Assert.Equal(
"2000", "2000",
processFactory.LastStartInfo.Environment[ processFactory.LastStartInfo.Environment[
WorkerProcessLauncher.WorkerPipeConnectAttemptTimeoutEnvironmentVariableName]); WorkerLaunchEnvironment.PipeConnectAttemptTimeoutMilliseconds]);
Assert.Equal( Assert.Equal(
"1500", "1500",
processFactory.LastStartInfo.Environment[ processFactory.LastStartInfo.Environment[
WorkerProcessLauncher.WorkerWriteCompletionWaitEnvironmentVariableName]); WorkerLaunchEnvironment.WriteCompletionWaitMilliseconds]);
// The worker sizes its outbound event queue from the launch environment; // 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 // the queue has no drop policy, so this capacity is the session's burst
// headroom rather than a throttle. // headroom rather than a throttle.
Assert.Equal( Assert.Equal(
"10000", "10000",
processFactory.LastStartInfo.Environment[ processFactory.LastStartInfo.Environment[
WorkerProcessLauncher.WorkerEventQueueCapacityEnvironmentVariableName]); WorkerLaunchEnvironment.EventQueueCapacity]);
// MxGateway:Alarms defaults reach the worker's alarm poll loop and its // MxGateway:Alarms defaults reach the worker's alarm poll loop and its
// GetXmlCurrentAlarms2 cap (which is also its truncation threshold) // GetXmlCurrentAlarms2 cap (which is also its truncation threshold)
// through the launch environment, not the command line. // through the launch environment, not the command line.
Assert.Equal( Assert.Equal(
"500", "500",
processFactory.LastStartInfo.Environment[ processFactory.LastStartInfo.Environment[
WorkerProcessLauncher.WorkerAlarmPollIntervalEnvironmentVariableName]); WorkerLaunchEnvironment.AlarmPollIntervalMilliseconds]);
Assert.Equal( Assert.Equal(
"1024", "1024",
processFactory.LastStartInfo.Environment[ processFactory.LastStartInfo.Environment[
WorkerProcessLauncher.WorkerMaxAlarmsPerFetchEnvironmentVariableName]); WorkerLaunchEnvironment.MaxAlarmsPerFetch]);
Assert.All(
RequiredLaunchEnvironmentVariables,
variable => Assert.True(
processFactory.LastStartInfo.Environment.ContainsKey(variable),
$"Worker launch environment is missing required variable '{variable}'."));
Assert.DoesNotContain(Nonce, handle.CommandLine.ToString(), StringComparison.Ordinal); Assert.DoesNotContain(Nonce, handle.CommandLine.ToString(), StringComparison.Ordinal);
Assert.DoesNotContain(Nonce, string.Join(" ", handle.CommandLine.Arguments), StringComparison.Ordinal); Assert.DoesNotContain(Nonce, string.Join(" ", handle.CommandLine.Arguments), StringComparison.Ordinal);
Assert.False(pipeReservation.DisposeCalled); Assert.False(pipeReservation.DisposeCalled);
@@ -94,7 +109,7 @@ public sealed class WorkerProcessLauncherTests
Assert.Equal( Assert.Equal(
"65536", "65536",
processFactory.LastStartInfo.Environment[ processFactory.LastStartInfo.Environment[
WorkerProcessLauncher.WorkerEventQueueCapacityEnvironmentVariableName]); WorkerLaunchEnvironment.EventQueueCapacity]);
} }
/// <summary>Verifies that a failed startup probe kills and disposes the worker process.</summary> /// <summary>Verifies that a failed startup probe kills and disposes the worker process.</summary>
@@ -137,7 +137,7 @@ public sealed class WorkerApplicationTests
private static MemoryWorkerEnvironment CreateEnvironment(string nonce) private static MemoryWorkerEnvironment CreateEnvironment(string nonce)
{ {
MemoryWorkerEnvironment environment = new(); MemoryWorkerEnvironment environment = new();
environment.Set(WorkerOptions.NonceEnvironmentVariableName, nonce); environment.Set(WorkerLaunchEnvironment.WorkerNonce, nonce);
return environment; return environment;
} }
@@ -115,7 +115,7 @@ public sealed class WorkerOptionsParserTests
private static MemoryWorkerEnvironment CreateEnvironment(string nonce) private static MemoryWorkerEnvironment CreateEnvironment(string nonce)
{ {
MemoryWorkerEnvironment environment = new(); MemoryWorkerEnvironment environment = new();
environment.Set(WorkerOptions.NonceEnvironmentVariableName, nonce); environment.Set(WorkerLaunchEnvironment.WorkerNonce, nonce);
return environment; return environment;
} }
} }
@@ -4,6 +4,7 @@ using System.Diagnostics;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Google.Protobuf.WellKnownTypes; using Google.Protobuf.WellKnownTypes;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Contracts.Proto; using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Worker.MxAccess; using ZB.MOM.WW.MxGateway.Worker.MxAccess;
@@ -609,11 +610,11 @@ public sealed class MxAccessEventQueueTests
int expected) int expected)
{ {
string? original = Environment.GetEnvironmentVariable( string? original = Environment.GetEnvironmentVariable(
MxAccessEventQueue.CapacityEnvironmentVariableName); WorkerLaunchEnvironment.EventQueueCapacity);
try try
{ {
Environment.SetEnvironmentVariable( Environment.SetEnvironmentVariable(
MxAccessEventQueue.CapacityEnvironmentVariableName, WorkerLaunchEnvironment.EventQueueCapacity,
environmentValue); environmentValue);
Assert.Equal(expected, MxAccessEventQueue.ResolveCapacity()); Assert.Equal(expected, MxAccessEventQueue.ResolveCapacity());
@@ -621,7 +622,7 @@ public sealed class MxAccessEventQueueTests
finally finally
{ {
Environment.SetEnvironmentVariable( Environment.SetEnvironmentVariable(
MxAccessEventQueue.CapacityEnvironmentVariableName, WorkerLaunchEnvironment.EventQueueCapacity,
original); original);
} }
} }
@@ -635,11 +636,11 @@ public sealed class MxAccessEventQueueTests
public void Constructor_WithResolvedCapacity_UsesEnvironmentValue() public void Constructor_WithResolvedCapacity_UsesEnvironmentValue()
{ {
string? original = Environment.GetEnvironmentVariable( string? original = Environment.GetEnvironmentVariable(
MxAccessEventQueue.CapacityEnvironmentVariableName); WorkerLaunchEnvironment.EventQueueCapacity);
try try
{ {
Environment.SetEnvironmentVariable( Environment.SetEnvironmentVariable(
MxAccessEventQueue.CapacityEnvironmentVariableName, WorkerLaunchEnvironment.EventQueueCapacity,
"2500"); "2500");
MxAccessEventQueue queue = new(MxAccessEventQueue.ResolveCapacity()); MxAccessEventQueue queue = new(MxAccessEventQueue.ResolveCapacity());
@@ -649,7 +650,7 @@ public sealed class MxAccessEventQueueTests
finally finally
{ {
Environment.SetEnvironmentVariable( Environment.SetEnvironmentVariable(
MxAccessEventQueue.CapacityEnvironmentVariableName, WorkerLaunchEnvironment.EventQueueCapacity,
original); original);
} }
} }
@@ -3,6 +3,7 @@ using System.Collections.Generic;
using System.Runtime.InteropServices; using System.Runtime.InteropServices;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Contracts.Proto; using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Worker.MxAccess; using ZB.MOM.WW.MxGateway.Worker.MxAccess;
using ZB.MOM.WW.MxGateway.Worker.Sta; using ZB.MOM.WW.MxGateway.Worker.Sta;
@@ -35,11 +36,11 @@ public sealed class MxAccessStaSessionTests
public void ResolveWriteCompletionTimeout_ParsesEnvironmentValue(string? rawValue, int? expectedMilliseconds) public void ResolveWriteCompletionTimeout_ParsesEnvironmentValue(string? rawValue, int? expectedMilliseconds)
{ {
string? original = Environment.GetEnvironmentVariable( string? original = Environment.GetEnvironmentVariable(
MxAccessStaSession.WriteCompletionWaitEnvironmentVariableName); WorkerLaunchEnvironment.WriteCompletionWaitMilliseconds);
try try
{ {
Environment.SetEnvironmentVariable( Environment.SetEnvironmentVariable(
MxAccessStaSession.WriteCompletionWaitEnvironmentVariableName, WorkerLaunchEnvironment.WriteCompletionWaitMilliseconds,
rawValue); rawValue);
TimeSpan resolved = MxAccessStaSession.ResolveWriteCompletionTimeout(); TimeSpan resolved = MxAccessStaSession.ResolveWriteCompletionTimeout();
@@ -52,7 +53,7 @@ public sealed class MxAccessStaSessionTests
finally finally
{ {
Environment.SetEnvironmentVariable( Environment.SetEnvironmentVariable(
MxAccessStaSession.WriteCompletionWaitEnvironmentVariableName, WorkerLaunchEnvironment.WriteCompletionWaitMilliseconds,
original); original);
} }
} }
@@ -83,11 +84,11 @@ public sealed class MxAccessStaSessionTests
int expectedMilliseconds) int expectedMilliseconds)
{ {
string? original = Environment.GetEnvironmentVariable( string? original = Environment.GetEnvironmentVariable(
MxAccessStaSession.AlarmPollIntervalEnvironmentVariableName); WorkerLaunchEnvironment.AlarmPollIntervalMilliseconds);
try try
{ {
Environment.SetEnvironmentVariable( Environment.SetEnvironmentVariable(
MxAccessStaSession.AlarmPollIntervalEnvironmentVariableName, WorkerLaunchEnvironment.AlarmPollIntervalMilliseconds,
environmentValue); environmentValue);
Assert.Equal( Assert.Equal(
@@ -97,7 +98,7 @@ public sealed class MxAccessStaSessionTests
finally finally
{ {
Environment.SetEnvironmentVariable( Environment.SetEnvironmentVariable(
MxAccessStaSession.AlarmPollIntervalEnvironmentVariableName, WorkerLaunchEnvironment.AlarmPollIntervalMilliseconds,
original); original);
} }
} }
@@ -3,6 +3,7 @@ using System.Collections.Generic;
using System.Linq; using System.Linq;
using System.Reflection; using System.Reflection;
using System.Threading; using System.Threading;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Worker.MxAccess; using ZB.MOM.WW.MxGateway.Worker.MxAccess;
namespace ZB.MOM.WW.MxGateway.Worker.Tests.MxAccess; namespace ZB.MOM.WW.MxGateway.Worker.Tests.MxAccess;
@@ -506,11 +507,11 @@ public sealed class WnWrapAlarmConsumerXmlTests
int expected) int expected)
{ {
string? original = Environment.GetEnvironmentVariable( string? original = Environment.GetEnvironmentVariable(
WnWrapAlarmConsumer.MaxAlarmsPerFetchEnvironmentVariableName); WorkerLaunchEnvironment.MaxAlarmsPerFetch);
try try
{ {
Environment.SetEnvironmentVariable( Environment.SetEnvironmentVariable(
WnWrapAlarmConsumer.MaxAlarmsPerFetchEnvironmentVariableName, WorkerLaunchEnvironment.MaxAlarmsPerFetch,
environmentValue); environmentValue);
Assert.Equal(expected, WnWrapAlarmConsumer.ResolveMaxAlarmsPerFetch()); Assert.Equal(expected, WnWrapAlarmConsumer.ResolveMaxAlarmsPerFetch());
@@ -518,7 +519,7 @@ public sealed class WnWrapAlarmConsumerXmlTests
finally finally
{ {
Environment.SetEnvironmentVariable( Environment.SetEnvironmentVariable(
WnWrapAlarmConsumer.MaxAlarmsPerFetchEnvironmentVariableName, WorkerLaunchEnvironment.MaxAlarmsPerFetch,
original); original);
} }
} }
@@ -1,10 +1,13 @@
namespace ZB.MOM.WW.MxGateway.Worker.Bootstrap; namespace ZB.MOM.WW.MxGateway.Worker.Bootstrap;
using ZB.MOM.WW.MxGateway.Contracts;
/// <summary>Worker bootstrap options passed via environment variables and named pipes.</summary> /// <summary>Worker bootstrap options passed via environment variables and named pipes.</summary>
public sealed class WorkerOptions public sealed class WorkerOptions
{ {
/// <summary>Environment variable name for the worker nonce.</summary> /// <summary>Environment variable name for the worker nonce.</summary>
public const string NonceEnvironmentVariableName = "MXGATEWAY_WORKER_NONCE"; [Obsolete("Use WorkerLaunchEnvironment.WorkerNonce.")]
public const string NonceEnvironmentVariableName = WorkerLaunchEnvironment.WorkerNonce;
/// <summary>Initializes worker options from a bootstrap handshake.</summary> /// <summary>Initializes worker options from a bootstrap handshake.</summary>
/// <param name="sessionId">Identifier of the session.</param> /// <param name="sessionId">Identifier of the session.</param>
@@ -75,7 +75,7 @@ public sealed class WorkerOptionsParser
[$"Unsupported protocol version '{protocolVersionText}'."]); [$"Unsupported protocol version '{protocolVersionText}'."]);
} }
string? nonce = _environment.GetEnvironmentVariable(WorkerOptions.NonceEnvironmentVariableName); string? nonce = _environment.GetEnvironmentVariable(WorkerLaunchEnvironment.WorkerNonce);
if (string.IsNullOrWhiteSpace(nonce)) if (string.IsNullOrWhiteSpace(nonce))
{ {
@@ -4,6 +4,7 @@ using System.IO;
using System.IO.Pipes; using System.IO.Pipes;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Worker.Bootstrap; using ZB.MOM.WW.MxGateway.Worker.Bootstrap;
using Polly; using Polly;
using Polly.Retry; using Polly.Retry;
@@ -22,8 +23,9 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
public const int DefaultConnectAttemptTimeoutMilliseconds = 2000; public const int DefaultConnectAttemptTimeoutMilliseconds = 2000;
/// <summary>Environment variable for overriding the per-attempt connection timeout.</summary> /// <summary>Environment variable for overriding the per-attempt connection timeout.</summary>
[Obsolete("Use WorkerLaunchEnvironment.PipeConnectAttemptTimeoutMilliseconds.")]
public const string ConnectAttemptTimeoutEnvironmentVariableName = public const string ConnectAttemptTimeoutEnvironmentVariableName =
"MXGATEWAY_WORKER_PIPE_CONNECT_ATTEMPT_TIMEOUT_MS"; WorkerLaunchEnvironment.PipeConnectAttemptTimeoutMilliseconds;
private readonly int _connectTimeoutMilliseconds; private readonly int _connectTimeoutMilliseconds;
private readonly int _connectAttemptTimeoutMilliseconds; private readonly int _connectAttemptTimeoutMilliseconds;
@@ -255,7 +257,8 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
private static int ResolveDefaultConnectAttemptTimeoutMilliseconds() private static int ResolveDefaultConnectAttemptTimeoutMilliseconds()
{ {
string? configuredValue = Environment.GetEnvironmentVariable(ConnectAttemptTimeoutEnvironmentVariableName); string? configuredValue = Environment.GetEnvironmentVariable(
WorkerLaunchEnvironment.PipeConnectAttemptTimeoutMilliseconds);
return int.TryParse(configuredValue, out int milliseconds) && milliseconds > 0 return int.TryParse(configuredValue, out int milliseconds) && milliseconds > 0
? milliseconds ? milliseconds
: DefaultConnectAttemptTimeoutMilliseconds; : DefaultConnectAttemptTimeoutMilliseconds;
@@ -5,6 +5,7 @@ using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Google.Protobuf; using Google.Protobuf;
using Google.Protobuf.WellKnownTypes; using Google.Protobuf.WellKnownTypes;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Contracts.Proto; using ZB.MOM.WW.MxGateway.Contracts.Proto;
namespace ZB.MOM.WW.MxGateway.Worker.MxAccess; namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
@@ -45,8 +46,6 @@ public sealed class MxAccessEventQueue
/// <see cref="DefaultCapacity"/> — a bad environment value must never /// <see cref="DefaultCapacity"/> — a bad environment value must never
/// keep the session from starting. /// keep the session from starting.
/// </summary> /// </summary>
internal const string CapacityEnvironmentVariableName = "MXGATEWAY_EVENT_QUEUE_CAPACITY";
/// <summary> /// <summary>
/// Floor on the resolved capacity. Mirrors the gateway-side /// Floor on the resolved capacity. Mirrors the gateway-side
/// <c>MxGateway:Worker:EventQueueCapacity</c> minimum so a value that /// <c>MxGateway:Worker:EventQueueCapacity</c> minimum so a value that
@@ -140,7 +139,7 @@ public sealed class MxAccessEventQueue
/// <returns>The capacity to construct the outbound event queue with.</returns> /// <returns>The capacity to construct the outbound event queue with.</returns>
internal static int ResolveCapacity() internal static int ResolveCapacity()
{ {
string? value = Environment.GetEnvironmentVariable(CapacityEnvironmentVariableName); string? value = Environment.GetEnvironmentVariable(WorkerLaunchEnvironment.EventQueueCapacity);
return int.TryParse( return int.TryParse(
value, value,
NumberStyles.Integer, NumberStyles.Integer,
@@ -3,6 +3,7 @@ using System.Collections.Generic;
using System.Diagnostics; using System.Diagnostics;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Contracts.Proto; using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Worker.Conversion; using ZB.MOM.WW.MxGateway.Worker.Conversion;
using ZB.MOM.WW.MxGateway.Worker.Sta; using ZB.MOM.WW.MxGateway.Worker.Sta;
@@ -16,17 +17,11 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
/// MxGateway:Worker:WriteCompletionWaitMilliseconds. 0 disables the /// MxGateway:Worker:WriteCompletionWaitMilliseconds. 0 disables the
/// write-completion wait (pure fire-and-forget replies). /// write-completion wait (pure fire-and-forget replies).
/// </summary> /// </summary>
internal const string WriteCompletionWaitEnvironmentVariableName =
"MXGATEWAY_WORKER_WRITE_COMPLETION_WAIT_MS";
/// <summary> /// <summary>
/// Environment variable the gateway's WorkerProcessLauncher sets from /// Environment variable the gateway's WorkerProcessLauncher sets from
/// MxGateway:Alarms:PollIntervalMilliseconds. A missing or invalid /// MxGateway:Alarms:PollIntervalMilliseconds. A missing or invalid
/// value falls back to <see cref="DefaultAlarmPollInterval"/>. /// value falls back to <see cref="DefaultAlarmPollInterval"/>.
/// </summary> /// </summary>
internal const string AlarmPollIntervalEnvironmentVariableName =
"MXGATEWAY_ALARM_POLL_INTERVAL_MS";
/// <summary> /// <summary>
/// Floor on the resolved alarm poll cadence. Mirrors the gateway-side /// Floor on the resolved alarm poll cadence. Mirrors the gateway-side
/// MxGateway:Alarms:PollIntervalMilliseconds minimum so a value that /// MxGateway:Alarms:PollIntervalMilliseconds minimum so a value that
@@ -211,7 +206,7 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
/// </summary> /// </summary>
internal static TimeSpan ResolveWriteCompletionTimeout() internal static TimeSpan ResolveWriteCompletionTimeout()
{ {
string? value = Environment.GetEnvironmentVariable(WriteCompletionWaitEnvironmentVariableName); string? value = Environment.GetEnvironmentVariable(WorkerLaunchEnvironment.WriteCompletionWaitMilliseconds);
return int.TryParse( return int.TryParse(
value, value,
System.Globalization.NumberStyles.Integer, System.Globalization.NumberStyles.Integer,
@@ -233,7 +228,7 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
/// <returns>The cadence the alarm poll loop waits between polls.</returns> /// <returns>The cadence the alarm poll loop waits between polls.</returns>
internal static TimeSpan ResolveAlarmPollInterval() internal static TimeSpan ResolveAlarmPollInterval()
{ {
string? value = Environment.GetEnvironmentVariable(AlarmPollIntervalEnvironmentVariableName); string? value = Environment.GetEnvironmentVariable(WorkerLaunchEnvironment.AlarmPollIntervalMilliseconds);
if (!int.TryParse( if (!int.TryParse(
value, value,
System.Globalization.NumberStyles.Integer, System.Globalization.NumberStyles.Integer,
@@ -5,6 +5,7 @@ using System.Globalization;
using System.Runtime.InteropServices; using System.Runtime.InteropServices;
using System.Xml; using System.Xml;
using WNWRAPCONSUMERLib; using WNWRAPCONSUMERLib;
using ZB.MOM.WW.MxGateway.Contracts;
namespace ZB.MOM.WW.MxGateway.Worker.MxAccess; namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
@@ -75,9 +76,6 @@ public sealed class WnWrapAlarmConsumer : IMxAccessAlarmConsumer
/// <see cref="DefaultMaxAlarmsPerFetch"/> — a bad environment value /// <see cref="DefaultMaxAlarmsPerFetch"/> — a bad environment value
/// must never keep the alarm consumer from starting. /// must never keep the alarm consumer from starting.
/// </summary> /// </summary>
internal const string MaxAlarmsPerFetchEnvironmentVariableName =
"MXGATEWAY_ALARM_MAX_ALARMS_PER_FETCH";
/// <summary> /// <summary>
/// Minimum gap between truncated-fetch warnings. A galaxy parked above /// Minimum gap between truncated-fetch warnings. A galaxy parked above
/// the cap truncates on *every* poll, so an unthrottled warning would /// the cap truncates on *every* poll, so an unthrottled warning would
@@ -156,7 +154,7 @@ public sealed class WnWrapAlarmConsumer : IMxAccessAlarmConsumer
/// <returns>The cap passed to <c>GetXmlCurrentAlarms2</c>.</returns> /// <returns>The cap passed to <c>GetXmlCurrentAlarms2</c>.</returns>
internal static int ResolveMaxAlarmsPerFetch() internal static int ResolveMaxAlarmsPerFetch()
{ {
string? value = Environment.GetEnvironmentVariable(MaxAlarmsPerFetchEnvironmentVariableName); string? value = Environment.GetEnvironmentVariable(WorkerLaunchEnvironment.MaxAlarmsPerFetch);
return int.TryParse( return int.TryParse(
value, value,
NumberStyles.Integer, NumberStyles.Integer,