dohertj2/mxaccessgw
1
0
Fork 0
Code Issues Pull Requests 2 Actions 29 Packages Projects Releases Wiki Activity

feat(ipc): negotiate worker frame max, add gRPC headroom, bound DrainEvents (IPC-02/03/04 gateway half)

Browse Source 
Proto foundation + gateway-side of the size/backpressure pass:

- IPC-02: add GatewayHello.max_frame_bytes (regen Generated/); gateway sends
  its negotiated worker-frame max in the handshake so the worker can adopt it
  instead of a hard-coded default. Worker read-half lands separately.
- IPC-03: give the pipe frame max envelope-overhead headroom above the public
  gRPC cap (WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes = 64 KiB;
  default Worker.MaxMessageBytes bumped to 16 MiB + reserve), cross-validate the
  headroom at startup, and pre-check command envelope size in WorkerClient so an
  oversized command fails only that correlation (ResourceExhausted) instead of
  faulting the whole session.
- IPC-04: reject DrainEvents max_events above a public ceiling in the request
  validator (worker per-reply cap lands with the worker half).

Docs: GatewayConfiguration, WorkerFrameProtocol, gateway.md.
Tests: headroom validation, DrainEvents bound, oversized-command per-command
failure (pipe-harness, verified on windev).
This commit is contained in:
Joseph Doherty
2026-07-09 08:49:59 -04:00
parent e1a505d662
commit c8b3a2281a
16 changed files with 362 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.MxGateway.Contracts/Generated/MxaccessWorker.cs
+97 -52
View File
@@ -44,63 +44,64 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
"Y2Nlc3Nfd29ya2VyLnYxLldvcmtlckV2ZW50SAASPwoQd29ya2VyX2hlYXJ0",
"YmVhdBgTIAEoCzIjLm14YWNjZXNzX3dvcmtlci52MS5Xb3JrZXJIZWFydGJl",
"YXRIABI3Cgx3b3JrZXJfZmF1bHQYFCABKAsyHy5teGFjY2Vzc193b3JrZXIu",
"djEuV29ya2VyRmF1bHRIAEIGCgRib2R5IloKDEdhdGV3YXlIZWxsbxIiChpz",
"djEuV29ya2VyRmF1bHRIAEIGCgRib2R5InMKDEdhdGV3YXlIZWxsbxIiChpz",
"dXBwb3J0ZWRfcHJvdG9jb2xfdmVyc2lvbhgBIAEoDRINCgVub25jZRgCIAEo",
"CRIXCg9nYXRld2F5X3ZlcnNpb24YAyABKAkiaQoLV29ya2VySGVsbG8SGAoQ",
"cHJvdG9jb2xfdmVyc2lvbhgBIAEoDRINCgVub25jZRgCIAEoCRIZChF3b3Jr",
"ZXJfcHJvY2Vzc19pZBgDIAEoBRIWCg53b3JrZXJfdmVyc2lvbhgEIAEoCSKO",
"AQoLV29ya2VyUmVhZHkSGQoRd29ya2VyX3Byb2Nlc3NfaWQYASABKAUSFwoP",
"bXhhY2Nlc3NfcHJvZ2lkGAIgASgJEhYKDm14YWNjZXNzX2Nsc2lkGAMgASgJ",
"EjMKD3JlYWR5X3RpbWVzdGFtcBgEIAEoCzIaLmdvb2dsZS5wcm90b2J1Zi5U",
"aW1lc3RhbXAidwoNV29ya2VyQ29tbWFuZBIvCgdjb21tYW5kGAEgASgLMh4u",
"bXhhY2Nlc3NfZ2F0ZXdheS52MS5NeENvbW1hbmQSNQoRZW5xdWV1ZV90aW1l",
"c3RhbXAYAiABKAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wIoEBChJX",
"b3JrZXJDb21tYW5kUmVwbHkSMgoFcmVwbHkYASABKAsyIy5teGFjY2Vzc19n",
"YXRld2F5LnYxLk14Q29tbWFuZFJlcGx5EjcKE2NvbXBsZXRlZF90aW1lc3Rh",
"bXAYAiABKAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wIh4KDFdvcmtl",
"ckNhbmNlbBIOCgZyZWFzb24YASABKAkiUQoOV29ya2VyU2h1dGRvd24SLwoM",
"Z3JhY2VfcGVyaW9kGAEgASgLMhkuZ29vZ2xlLnByb3RvYnVmLkR1cmF0aW9u",
"Eg4KBnJlYXNvbhgCIAEoCSJIChFXb3JrZXJTaHV0ZG93bkFjaxIzCgZzdGF0",
"dXMYASABKAsyIy5teGFjY2Vzc19nYXRld2F5LnYxLlByb3RvY29sU3RhdHVz",
"IjoKC1dvcmtlckV2ZW50EisKBWV2ZW50GAEgASgLMhwubXhhY2Nlc3NfZ2F0",
"ZXdheS52MS5NeEV2ZW50IqUCCg9Xb3JrZXJIZWFydGJlYXQSGQoRd29ya2Vy",
"X3Byb2Nlc3NfaWQYASABKAUSLgoFc3RhdGUYAiABKA4yHy5teGFjY2Vzc193",
"b3JrZXIudjEuV29ya2VyU3RhdGUSPwobbGFzdF9zdGFfYWN0aXZpdHlfdGlt",
"ZXN0YW1wGAMgASgLMhouZ29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcBIdChVw",
"ZW5kaW5nX2NvbW1hbmRfY291bnQYBCABKA0SIgoab3V0Ym91bmRfZXZlbnRf",
"cXVldWVfZGVwdGgYBSABKA0SGwoTbGFzdF9ldmVudF9zZXF1ZW5jZRgGIAEo",
"BBImCh5jdXJyZW50X2NvbW1hbmRfY29ycmVsYXRpb25faWQYByABKAki9AEK",
"C1dvcmtlckZhdWx0EjkKCGNhdGVnb3J5GAEgASgOMicubXhhY2Nlc3Nfd29y",
"a2VyLnYxLldvcmtlckZhdWx0Q2F0ZWdvcnkSFgoOY29tbWFuZF9tZXRob2QY",
"AiABKAkSFAoHaHJlc3VsdBgDIAEoBUgAiAEBEhYKDmV4Y2VwdGlvbl90eXBl",
"GAQgASgJEhoKEmRpYWdub3N0aWNfbWVzc2FnZRgFIAEoCRI8Cg9wcm90b2Nv",
"bF9zdGF0dXMYBiABKAsyIy5teGFjY2Vzc19nYXRld2F5LnYxLlByb3RvY29s",
"U3RhdHVzQgoKCF9ocmVzdWx0KpcCCgtXb3JrZXJTdGF0ZRIcChhXT1JLRVJf",
"U1RBVEVfVU5TUEVDSUZJRUQQABIZChVXT1JLRVJfU1RBVEVfU1RBUlRJTkcQ",
"ARIcChhXT1JLRVJfU1RBVEVfSEFORFNIQUtJTkcQAhIhCh1XT1JLRVJfU1RB",
"VEVfSU5JVElBTElaSU5HX1NUQRADEhYKEldPUktFUl9TVEFURV9SRUFEWRAE",
"EiIKHldPUktFUl9TVEFURV9FWEVDVVRJTkdfQ09NTUFORBAFEh4KGldPUktF",
"Ul9TVEFURV9TSFVUVElOR19ET1dOEAYSGAoUV09SS0VSX1NUQVRFX1NUT1BQ",
"RUQQBxIYChRXT1JLRVJfU1RBVEVfRkFVTFRFRBAIKscEChNXb3JrZXJGYXVs",
"dENhdGVnb3J5EiUKIVdPUktFUl9GQVVMVF9DQVRFR09SWV9VTlNQRUNJRklF",
"RBAAEisKJ1dPUktFUl9GQVVMVF9DQVRFR09SWV9JTlZBTElEX0FSR1VNRU5U",
"UxABEjcKM1dPUktFUl9GQVVMVF9DQVRFR09SWV9HQVRFV0FZX0FVVEhFTlRJ",
"Q0FUSU9OX0ZBSUxFRBACEisKJ1dPUktFUl9GQVVMVF9DQVRFR09SWV9QUk9U",
"T0NPTF9NSVNNQVRDSBADEiwKKFdPUktFUl9GQVVMVF9DQVRFR09SWV9QUk9U",
"T0NPTF9WSU9MQVRJT04QBBIrCidXT1JLRVJfRkFVTFRfQ0FURUdPUllfUElQ",
"RV9ESVNDT05ORUNURUQQBRIyCi5XT1JLRVJfRkFVTFRfQ0FURUdPUllfTVhB",
"Q0NFU1NfQ1JFQVRJT05fRkFJTEVEEAYSMQotV09SS0VSX0ZBVUxUX0NBVEVH",
"T1JZX01YQUNDRVNTX0NPTU1BTkRfRkFJTEVEEAcSOgo2V09SS0VSX0ZBVUxU",
"X0NBVEVHT1JZX01YQUNDRVNTX0VWRU5UX0NPTlZFUlNJT05fRkFJTEVEEAgS",
"IgoeV09SS0VSX0ZBVUxUX0NBVEVHT1JZX1NUQV9IVU5HEAkSKAokV09SS0VS",
"X0ZBVUxUX0NBVEVHT1JZX1FVRVVFX09WRVJGTE9XEAoSKgomV09SS0VSX0ZB",
"VUxUX0NBVEVHT1JZX1NIVVRET1dOX1RJTUVPVVQQC0ImqgIjWkIuTU9NLldX",
"Lk14R2F0ZXdheS5Db250cmFjdHMuUHJvdG9iBnByb3RvMw=="));
"CRIXCg9nYXRld2F5X3ZlcnNpb24YAyABKAkSFwoPbWF4X2ZyYW1lX2J5dGVz",
"GAQgASgNImkKC1dvcmtlckhlbGxvEhgKEHByb3RvY29sX3ZlcnNpb24YASAB",
"KA0SDQoFbm9uY2UYAiABKAkSGQoRd29ya2VyX3Byb2Nlc3NfaWQYAyABKAUS",
"FgoOd29ya2VyX3ZlcnNpb24YBCABKAkijgEKC1dvcmtlclJlYWR5EhkKEXdv",
"cmtlcl9wcm9jZXNzX2lkGAEgASgFEhcKD214YWNjZXNzX3Byb2dpZBgCIAEo",
"CRIWCg5teGFjY2Vzc19jbHNpZBgDIAEoCRIzCg9yZWFkeV90aW1lc3RhbXAY",
"BCABKAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wIncKDVdvcmtlckNv",
"bW1hbmQSLwoHY29tbWFuZBgBIAEoCzIeLm14YWNjZXNzX2dhdGV3YXkudjEu",
"TXhDb21tYW5kEjUKEWVucXVldWVfdGltZXN0YW1wGAIgASgLMhouZ29vZ2xl",
"LnByb3RvYnVmLlRpbWVzdGFtcCKBAQoSV29ya2VyQ29tbWFuZFJlcGx5EjIK",
"BXJlcGx5GAEgASgLMiMubXhhY2Nlc3NfZ2F0ZXdheS52MS5NeENvbW1hbmRS",
"ZXBseRI3ChNjb21wbGV0ZWRfdGltZXN0YW1wGAIgASgLMhouZ29vZ2xlLnBy",
"b3RvYnVmLlRpbWVzdGFtcCIeCgxXb3JrZXJDYW5jZWwSDgoGcmVhc29uGAEg",
"ASgJIlEKDldvcmtlclNodXRkb3duEi8KDGdyYWNlX3BlcmlvZBgBIAEoCzIZ",
"Lmdvb2dsZS5wcm90b2J1Zi5EdXJhdGlvbhIOCgZyZWFzb24YAiABKAkiSAoR",
"V29ya2VyU2h1dGRvd25BY2sSMwoGc3RhdHVzGAEgASgLMiMubXhhY2Nlc3Nf",
"Z2F0ZXdheS52MS5Qcm90b2NvbFN0YXR1cyI6CgtXb3JrZXJFdmVudBIrCgVl",
"dmVudBgBIAEoCzIcLm14YWNjZXNzX2dhdGV3YXkudjEuTXhFdmVudCKlAgoP",
"V29ya2VySGVhcnRiZWF0EhkKEXdvcmtlcl9wcm9jZXNzX2lkGAEgASgFEi4K",
"BXN0YXRlGAIgASgOMh8ubXhhY2Nlc3Nfd29ya2VyLnYxLldvcmtlclN0YXRl",
"Ej8KG2xhc3Rfc3RhX2FjdGl2aXR5X3RpbWVzdGFtcBgDIAEoCzIaLmdvb2ds",
"ZS5wcm90b2J1Zi5UaW1lc3RhbXASHQoVcGVuZGluZ19jb21tYW5kX2NvdW50",
"GAQgASgNEiIKGm91dGJvdW5kX2V2ZW50X3F1ZXVlX2RlcHRoGAUgASgNEhsK",
"E2xhc3RfZXZlbnRfc2VxdWVuY2UYBiABKAQSJgoeY3VycmVudF9jb21tYW5k",
"X2NvcnJlbGF0aW9uX2lkGAcgASgJIvQBCgtXb3JrZXJGYXVsdBI5CghjYXRl",
"Z29yeRgBIAEoDjInLm14YWNjZXNzX3dvcmtlci52MS5Xb3JrZXJGYXVsdENh",
"dGVnb3J5EhYKDmNvbW1hbmRfbWV0aG9kGAIgASgJEhQKB2hyZXN1bHQYAyAB",
"KAVIAIgBARIWCg5leGNlcHRpb25fdHlwZRgEIAEoCRIaChJkaWFnbm9zdGlj",
"X21lc3NhZ2UYBSABKAkSPAoPcHJvdG9jb2xfc3RhdHVzGAYgASgLMiMubXhh",
"Y2Nlc3NfZ2F0ZXdheS52MS5Qcm90b2NvbFN0YXR1c0IKCghfaHJlc3VsdCqX",
"AgoLV29ya2VyU3RhdGUSHAoYV09SS0VSX1NUQVRFX1VOU1BFQ0lGSUVEEAAS",
"GQoVV09SS0VSX1NUQVRFX1NUQVJUSU5HEAESHAoYV09SS0VSX1NUQVRFX0hB",
"TkRTSEFLSU5HEAISIQodV09SS0VSX1NUQVRFX0lOSVRJQUxJWklOR19TVEEQ",
"AxIWChJXT1JLRVJfU1RBVEVfUkVBRFkQBBIiCh5XT1JLRVJfU1RBVEVfRVhF",
"Q1VUSU5HX0NPTU1BTkQQBRIeChpXT1JLRVJfU1RBVEVfU0hVVFRJTkdfRE9X",
"ThAGEhgKFFdPUktFUl9TVEFURV9TVE9QUEVEEAcSGAoUV09SS0VSX1NUQVRF",
"X0ZBVUxURUQQCCrHBAoTV29ya2VyRmF1bHRDYXRlZ29yeRIlCiFXT1JLRVJf",
"RkFVTFRfQ0FURUdPUllfVU5TUEVDSUZJRUQQABIrCidXT1JLRVJfRkFVTFRf",
"Q0FURUdPUllfSU5WQUxJRF9BUkdVTUVOVFMQARI3CjNXT1JLRVJfRkFVTFRf",
"Q0FURUdPUllfR0FURVdBWV9BVVRIRU5USUNBVElPTl9GQUlMRUQQAhIrCidX",
"T1JLRVJfRkFVTFRfQ0FURUdPUllfUFJPVE9DT0xfTUlTTUFUQ0gQAxIsCihX",
"T1JLRVJfRkFVTFRfQ0FURUdPUllfUFJPVE9DT0xfVklPTEFUSU9OEAQSKwon",
"V09SS0VSX0ZBVUxUX0NBVEVHT1JZX1BJUEVfRElTQ09OTkVDVEVEEAUSMgou",
"V09SS0VSX0ZBVUxUX0NBVEVHT1JZX01YQUNDRVNTX0NSRUFUSU9OX0ZBSUxF",
"RBAGEjEKLVdPUktFUl9GQVVMVF9DQVRFR09SWV9NWEFDQ0VTU19DT01NQU5E",
"X0ZBSUxFRBAHEjoKNldPUktFUl9GQVVMVF9DQVRFR09SWV9NWEFDQ0VTU19F",
"VkVOVF9DT05WRVJTSU9OX0ZBSUxFRBAIEiIKHldPUktFUl9GQVVMVF9DQVRF",
"R09SWV9TVEFfSFVORxAJEigKJFdPUktFUl9GQVVMVF9DQVRFR09SWV9RVUVV",
"RV9PVkVSRkxPVxAKEioKJldPUktFUl9GQVVMVF9DQVRFR09SWV9TSFVURE9X",
"Tl9USU1FT1VUEAtCJqoCI1pCLk1PTS5XVy5NeEdhdGV3YXkuQ29udHJhY3Rz",
"LlByb3RvYgZwcm90bzM="));
descriptor = pbr::FileDescriptor.FromGeneratedCode(descriptorData,
new pbr::FileDescriptor[] { global::Google.Protobuf.WellKnownTypes.DurationReflection.Descriptor, global::Google.Protobuf.WellKnownTypes.TimestampReflection.Descriptor, global::ZB.MOM.WW.MxGateway.Contracts.Proto.MxaccessGatewayReflection.Descriptor, },
new pbr::GeneratedClrTypeInfo(new[] {typeof(global::ZB.MOM.WW.MxGateway.Contracts.Proto.WorkerState), typeof(global::ZB.MOM.WW.MxGateway.Contracts.Proto.WorkerFaultCategory), }, null, new pbr::GeneratedClrTypeInfo[] {
new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.MxGateway.Contracts.Proto.WorkerEnvelope), global::ZB.MOM.WW.MxGateway.Contracts.Proto.WorkerEnvelope.Parser, new[]{ "ProtocolVersion", "SessionId", "Sequence", "CorrelationId", "GatewayHello", "WorkerHello", "WorkerReady", "WorkerCommand", "WorkerCommandReply", "WorkerCancel", "WorkerShutdown", "WorkerShutdownAck", "WorkerEvent", "WorkerHeartbeat", "WorkerFault" }, new[]{ "Body" }, null, null, null),
new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.MxGateway.Contracts.Proto.GatewayHello), global::ZB.MOM.WW.MxGateway.Contracts.Proto.GatewayHello.Parser, new[]{ "SupportedProtocolVersion", "Nonce", "GatewayVersion" }, null, null, null, null),
new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.MxGateway.Contracts.Proto.GatewayHello), global::ZB.MOM.WW.MxGateway.Contracts.Proto.GatewayHello.Parser, new[]{ "SupportedProtocolVersion", "Nonce", "GatewayVersion", "MaxFrameBytes" }, null, null, null, null),
new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.MxGateway.Contracts.Proto.WorkerHello), global::ZB.MOM.WW.MxGateway.Contracts.Proto.WorkerHello.Parser, new[]{ "ProtocolVersion", "Nonce", "WorkerProcessId", "WorkerVersion" }, null, null, null, null),
new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.MxGateway.Contracts.Proto.WorkerReady), global::ZB.MOM.WW.MxGateway.Contracts.Proto.WorkerReady.Parser, new[]{ "WorkerProcessId", "MxaccessProgid", "MxaccessClsid", "ReadyTimestamp" }, null, null, null, null),
new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.MxGateway.Contracts.Proto.WorkerCommand), global::ZB.MOM.WW.MxGateway.Contracts.Proto.WorkerCommand.Parser, new[]{ "Command", "EnqueueTimestamp" }, null, null, null, null),
@@ -1108,6 +1109,7 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
supportedProtocolVersion_ = other.supportedProtocolVersion_;
nonce_ = other.nonce_;
gatewayVersion_ = other.gatewayVersion_;
maxFrameBytes_ = other.maxFrameBytes_;
_unknownFields = pb::UnknownFieldSet.Clone(other._unknownFields);
}
@@ -1153,6 +1155,25 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
}
}
/// <summary>Field number for the "max_frame_bytes" field.</summary>
public const int MaxFrameBytesFieldNumber = 4;
private uint maxFrameBytes_;
/// <summary>
/// Maximum worker-frame payload size, in bytes, negotiated by the gateway from its
/// configured pipe limit. The worker adopts this as its frame-protocol MaxMessageBytes
/// instead of a hard-coded default; 0 (an older gateway that never set the field) means
/// "use the worker's built-in default". Sits above the public gRPC cap by an
/// envelope-overhead margin so an accepted gRPC payload always fits one worker frame.
/// </summary>
[global::System.Diagnostics.DebuggerNonUserCodeAttribute]
[global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
public uint MaxFrameBytes {
get { return maxFrameBytes_; }
set {
maxFrameBytes_ = value;
}
}
[global::System.Diagnostics.DebuggerNonUserCodeAttribute]
[global::System.CodeDom.Compiler.GeneratedCode("protoc", null)]
public override bool Equals(object other) {
@@ -1171,6 +1192,7 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
if (SupportedProtocolVersion != other.SupportedProtocolVersion) return false;
if (Nonce != other.Nonce) return false;
if (GatewayVersion != other.GatewayVersion) return false;
if (MaxFrameBytes != other.MaxFrameBytes) return false;
return Equals(_unknownFields, other._unknownFields);
}
@@ -1181,6 +1203,7 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
if (SupportedProtocolVersion != 0) hash ^= SupportedProtocolVersion.GetHashCode();
if (Nonce.Length != 0) hash ^= Nonce.GetHashCode();
if (GatewayVersion.Length != 0) hash ^= GatewayVersion.GetHashCode();
if (MaxFrameBytes != 0) hash ^= MaxFrameBytes.GetHashCode();
if (_unknownFields != null) {
hash ^= _unknownFields.GetHashCode();
}
@@ -1211,6 +1234,10 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
output.WriteRawTag(26);
output.WriteString(GatewayVersion);
}
if (MaxFrameBytes != 0) {
output.WriteRawTag(32);
output.WriteUInt32(MaxFrameBytes);
}
if (_unknownFields != null) {
_unknownFields.WriteTo(output);
}
@@ -1233,6 +1260,10 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
output.WriteRawTag(26);
output.WriteString(GatewayVersion);
}
if (MaxFrameBytes != 0) {
output.WriteRawTag(32);
output.WriteUInt32(MaxFrameBytes);
}
if (_unknownFields != null) {
_unknownFields.WriteTo(ref output);
}
@@ -1252,6 +1283,9 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
if (GatewayVersion.Length != 0) {
size += 1 + pb::CodedOutputStream.ComputeStringSize(GatewayVersion);
}
if (MaxFrameBytes != 0) {
size += 1 + pb::CodedOutputStream.ComputeUInt32Size(MaxFrameBytes);
}
if (_unknownFields != null) {
size += _unknownFields.CalculateSize();
}
@@ -1273,6 +1307,9 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
if (other.GatewayVersion.Length != 0) {
GatewayVersion = other.GatewayVersion;
}
if (other.MaxFrameBytes != 0) {
MaxFrameBytes = other.MaxFrameBytes;
}
_unknownFields = pb::UnknownFieldSet.MergeFrom(_unknownFields, other._unknownFields);
}
@@ -1304,6 +1341,10 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
GatewayVersion = input.ReadString();
break;
}
case 32: {
MaxFrameBytes = input.ReadUInt32();
break;
}
}
}
#endif
@@ -1335,6 +1376,10 @@ namespace ZB.MOM.WW.MxGateway.Contracts.Proto {
GatewayVersion = input.ReadString();
break;
}
case 32: {
MaxFrameBytes = input.ReadUInt32();
break;
}
}
}
}
src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_worker.proto
+6
View File
@@ -42,6 +42,12 @@ message GatewayHello {
uint32 supported_protocol_version = 1;
string nonce = 2;
string gateway_version = 3;
// Maximum worker-frame payload size, in bytes, negotiated by the gateway from its
// configured pipe limit. The worker adopts this as its frame-protocol MaxMessageBytes
// instead of a hard-coded default; 0 (an older gateway that never set the field) means
// "use the worker's built-in default". Sits above the public gRPC cap by an
// envelope-overhead margin so an accepted gRPC payload always fits one worker frame.
uint32 max_frame_bytes = 4;
}
message WorkerHello {
src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs
+31
View File
@@ -1,6 +1,7 @@
using ZB.MOM.WW.Auth.Abstractions.Ldap;
using ZB.MOM.WW.Configuration;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Server.Workers;
namespace ZB.MOM.WW.MxGateway.Server.Configuration;
@@ -46,6 +47,7 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase<GatewayOption
ValidateEvents(options.Events, builder);
ValidateDashboard(options.Dashboard, builder, _isProduction);
ValidateProtocol(options.Protocol, builder);
ValidateFrameSizeHeadroom(options.Worker, options.Protocol, builder);
ValidateAlarms(options.Alarms, builder);
ValidateTls(options.Tls, builder);
ValidateSecurity(options.Security, builder);
@@ -466,6 +468,35 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase<GatewayOption
}
}
// The worker-frame (pipe) maximum must stay above the public gRPC cap by the envelope-overhead
// reserve, otherwise a maximally-sized accepted gRPC payload does not fit one worker frame once
// wrapped in a WorkerEnvelope and the outbound write faults the whole session (IPC-03). Fail fast
// at startup rather than mid-traffic. Only checked when both knobs are themselves in range so the
// message is not doubled up with the individual range errors.
private static void ValidateFrameSizeHeadroom(
WorkerOptions worker,
ProtocolOptions protocol,
ValidationBuilder builder)
{
bool workerInRange = worker.MaxMessageBytes is >= MinimumMaxMessageBytes and <= MaximumMaxMessageBytes;
bool grpcInRange = protocol.MaxGrpcMessageBytes is >= MinimumMaxMessageBytes and <= MaximumMaxMessageBytes;
if (!workerInRange || !grpcInRange)
{
return;
}
long requiredWorkerMax =
(long)protocol.MaxGrpcMessageBytes + WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes;
if (worker.MaxMessageBytes < requiredWorkerMax)
{
builder.Add(
$"MxGateway:Worker:MaxMessageBytes ({worker.MaxMessageBytes}) must be at least "
+ $"MxGateway:Protocol:MaxGrpcMessageBytes ({protocol.MaxGrpcMessageBytes}) plus the "
+ $"{WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes}-byte worker-frame envelope reserve "
+ $"(>= {requiredWorkerMax}).");
}
}
private static void AddIfBlank(string? value, string message, ValidationBuilder builder)
{
builder.RequireThat(!string.IsNullOrWhiteSpace(value), message);
src/ZB.MOM.WW.MxGateway.Server/Configuration/WorkerOptions.cs
+10 -2
View File
@@ -33,6 +33,14 @@ public sealed class WorkerOptions
/// <summary>The grace period in seconds after a heartbeat before considering the worker unresponsive.</summary>
public int HeartbeatGraceSeconds { get; init; } = 15;
/// <summary>The maximum message size in bytes for IPC communication.</summary>
public int MaxMessageBytes { get; init; } = 16 * 1024 * 1024;
/// <summary>
/// The maximum worker-frame (pipe) message size in bytes. Must stay at least
/// <see cref="Workers.WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes"/> above
/// <see cref="ProtocolOptions.MaxGrpcMessageBytes"/> so a maximally-sized accepted gRPC payload
/// still fits one worker frame (IPC-03); the gateway conveys this value to the worker in the
/// handshake (<c>GatewayHello.max_frame_bytes</c>, IPC-02). Default is the 16 MB public gRPC cap
/// plus that reserve.
/// </summary>
public int MaxMessageBytes { get; init; } =
(16 * 1024 * 1024) + Workers.WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes;
}
src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs
+1
View File
@@ -946,6 +946,7 @@ public sealed class MxAccessGatewayService(
WorkerClientErrorCode.GatewayShutdown => StatusCode.Cancelled,
WorkerClientErrorCode.InvalidState => StatusCode.FailedPrecondition,
WorkerClientErrorCode.ProtocolViolation => StatusCode.Internal,
WorkerClientErrorCode.CommandTooLarge => StatusCode.ResourceExhausted,
_ => StatusCode.Unavailable,
};
src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGrpcRequestValidator.cs
+15
View File
@@ -5,6 +5,13 @@ namespace ZB.MOM.WW.MxGateway.Server.Grpc;
public sealed class MxAccessGrpcRequestValidator
{
// Upper bound on a single DrainEvents request. DrainEvents is a diagnostics RPC that returns
// buffered events in one non-streaming reply, so an unbounded max_events could pack the whole
// queue into a session-killing frame (IPC-04). The worker independently caps each reply at its
// own MaxDrainEventsPerReply; this public bound rejects an obviously-abusive request loudly at
// the boundary. max_events = 0 is allowed and means "the worker's default batch cap".
private const uint MaxDrainEventsPerRequest = 10_000;
/// <summary>Validates an open session request.</summary>
/// <param name="request">The request to validate.</param>
public void ValidateOpenSession(OpenSessionRequest request)
@@ -69,6 +76,14 @@ public sealed class MxAccessGrpcRequestValidator
throw InvalidArgument(
$"Command kind {command.Kind} requires payload {expectedPayload} but received {command.PayloadCase}.");
}
// The payload case now matches the kind, so command.DrainEvents is non-null here.
if (command.Kind is MxCommandKind.DrainEvents && command.DrainEvents.MaxEvents > MaxDrainEventsPerRequest)
{
throw InvalidArgument(
$"DrainEvents max_events ({command.DrainEvents.MaxEvents}) must not exceed {MaxDrainEventsPerRequest}; "
+ "use 0 to request the worker default batch cap.");
}
}
private static MxCommand.PayloadOneofCase ExpectedPayload(MxCommandKind kind)
src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs
+21 -1
View File
@@ -187,7 +187,23 @@ public sealed class WorkerClient : IWorkerClient
try
{
await EnqueueAsync(CreateCommandEnvelope(correlationId, command), cancellationToken).ConfigureAwait(false);
WorkerEnvelope commandEnvelope = CreateCommandEnvelope(correlationId, command);
// Reject an oversized command at the enqueue boundary so only this correlation fails
// (ResourceExhausted) rather than the frame reaching the write loop and faulting the whole
// session (IPC-03). Command envelopes are the only gateway-authored outbound payload whose
// size the caller controls; checking here keeps a MessageTooLarge in the write loop a
// genuine desync signal.
int envelopeSize = commandEnvelope.CalculateSize();
if (envelopeSize > _connection.FrameOptions.MaxMessageBytes)
{
throw new WorkerClientException(
WorkerClientErrorCode.CommandTooLarge,
$"Worker command {method} serializes to {envelopeSize} bytes, exceeding the negotiated "
+ $"worker-frame maximum of {_connection.FrameOptions.MaxMessageBytes} bytes.");
}
await EnqueueAsync(commandEnvelope, cancellationToken).ConfigureAwait(false);
using CancellationTokenSource timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
Task timeoutTask = Task.Delay(timeout, timeoutCts.Token);
Task<WorkerCommandReply> replyTask = pendingCommand.Task;
@@ -910,6 +926,10 @@ public sealed class WorkerClient : IWorkerClient
SupportedProtocolVersion = _connection.FrameOptions.ProtocolVersion,
Nonce = _connection.Nonce,
GatewayVersion = typeof(GatewayContractInfo).Assembly.GetName().Version?.ToString() ?? GatewayVersionFallback,
// Convey the negotiated worker-frame maximum so the worker adopts it instead of a
// hard-coded default (IPC-02). Sits above the public gRPC cap by the envelope reserve.
MaxFrameBytes = (uint)_connection.FrameOptions.MaxMessageBytes,
});
}
src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClientErrorCode.cs
+5
View File
@@ -12,4 +12,9 @@ public enum WorkerClientErrorCode
GatewayShutdown,
WriteFailed,
PendingCommandLimitExceeded,
// The serialized command envelope exceeds the negotiated worker-frame maximum. Rejected at the
// enqueue boundary so only the offending command fails (mapped to ResourceExhausted) instead of
// the oversized frame reaching the write loop and faulting the whole session (IPC-03).
CommandTooLarge,
}
src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerFrameProtocolOptions.cs
+9
View File
@@ -10,6 +10,15 @@ public sealed class WorkerFrameProtocolOptions
/// <summary>Default maximum message size in bytes (16 MB).</summary>
public const int DefaultMaxMessageBytes = 16 * 1024 * 1024;
/// <summary>
/// Byte margin the worker-frame (pipe) maximum must keep above the public gRPC message cap so a
/// gRPC payload accepted at the public boundary always fits inside one worker frame once wrapped
/// in a <c>WorkerEnvelope</c> (correlation id, timestamps, oneof framing). Without this headroom
/// the pipe max equals the gRPC max and a maximally-sized accepted request faults the whole
/// session on the outbound write (IPC-03). 64 KiB is far larger than the fixed envelope overhead.
/// </summary>
public const int EnvelopeOverheadReserveBytes = 64 * 1024;
/// <summary>
/// Initializes worker frame protocol options with a session ID.
/// </summary>
src/ZB.MOM.WW.MxGateway.Server/appsettings.json
+1 -1
View File
@@ -38,7 +38,7 @@
"ShutdownTimeoutSeconds": 10,
"HeartbeatIntervalSeconds": 5,
"HeartbeatGraceSeconds": 15,
"MaxMessageBytes": 16777216
"MaxMessageBytes": 16842752
},
"Sessions": {
"DefaultCommandTimeoutSeconds": 30,
src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs
+46
View File
@@ -771,4 +771,50 @@ public sealed class GatewayOptionsValidatorTests
Assert.True(result.Failed);
Assert.Contains(result.Failures!, f => f.Contains("ApiKeyFailureTrackedPeers"));
}
private static GatewayOptions WithWorkerAndProtocol(WorkerOptions worker, ProtocolOptions protocol)
{
GatewayOptions source = ValidOptions();
return new GatewayOptions
{
Authentication = source.Authentication,
Ldap = source.Ldap,
Worker = worker,
Sessions = source.Sessions,
Events = source.Events,
Dashboard = source.Dashboard,
Protocol = protocol,
Alarms = source.Alarms,
Tls = source.Tls,
};
}
/// <summary>
/// Verifies the default worker-frame maximum keeps the required envelope-overhead reserve above
/// the default public gRPC cap, so a stock configuration passes the IPC-03 headroom check.
/// </summary>
[Fact]
public void Validate_Succeeds_WhenWorkerFrameMaxHasEnvelopeHeadroom()
{
ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, ValidOptions());
Assert.True(result.Succeeded);
}
/// <summary>
/// Verifies that a worker-frame maximum equal to the gRPC cap (zero headroom) fails validation:
/// a maximally-sized accepted gRPC payload would not fit one worker frame once wrapped in a
/// WorkerEnvelope, faulting the whole session on the outbound write (IPC-03).
/// </summary>
[Fact]
public void Validate_Fails_WhenWorkerFrameMaxEqualsGrpcMaxWithoutHeadroom()
{
const int grpcMax = 16 * 1024 * 1024;
ValidateOptionsResult result = new GatewayOptionsValidator().Validate(
null,
WithWorkerAndProtocol(
new WorkerOptions { MaxMessageBytes = grpcMax },
new ProtocolOptions { MaxGrpcMessageBytes = grpcMax }));
Assert.True(result.Failed);
Assert.Contains(result.Failures!, f => f.Contains("MaxMessageBytes") && f.Contains("reserve"));
}
}
src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGrpcRequestValidatorTests.cs
+45
View File
@@ -0,0 +1,45 @@
using Grpc.Core;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Grpc;
namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Grpc;
public sealed class MxAccessGrpcRequestValidatorTests
{
private static MxCommandRequest DrainRequest(uint maxEvents) => new()
{
SessionId = "session-1",
Command = new MxCommand
{
Kind = MxCommandKind.DrainEvents,
DrainEvents = new DrainEventsCommand { MaxEvents = maxEvents },
},
};
/// <summary>
/// Verifies a DrainEvents request within the per-request ceiling passes validation, including the
/// <c>max_events = 0</c> "worker default cap" sentinel (IPC-04).
/// </summary>
[Theory]
[InlineData(0u)]
[InlineData(1u)]
[InlineData(10_000u)]
public void ValidateInvoke_AllowsDrainEvents_WithinCeiling(uint maxEvents)
{
MxAccessGrpcRequestValidator validator = new();
validator.ValidateInvoke(DrainRequest(maxEvents));
}
/// <summary>
/// Verifies a DrainEvents request above the per-request ceiling is rejected with InvalidArgument
/// so one accepted request cannot pack an unbounded reply frame (IPC-04).
/// </summary>
[Fact]
public void ValidateInvoke_RejectsDrainEvents_AboveCeiling()
{
MxAccessGrpcRequestValidator validator = new();
RpcException exception = Assert.Throws<RpcException>(() => validator.ValidateInvoke(DrainRequest(10_001)));
Assert.Equal(StatusCode.InvalidArgument, exception.StatusCode);
Assert.Contains("max_events", exception.Status.Detail, StringComparison.Ordinal);
}
}
src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs
+52 -2
View File
@@ -56,6 +56,52 @@ public sealed class WorkerClientTests
Assert.Equal(MxCommandKind.Ping, reply.Reply.Kind);
}
/// <summary>
/// Verifies that a command whose serialized envelope exceeds the negotiated worker-frame maximum
/// fails only that command with <see cref="WorkerClientErrorCode.CommandTooLarge"/> at the enqueue
/// boundary, leaving the client ready for subsequent commands (IPC-03). Without the pre-check the
/// oversized frame would reach the write loop and fault the whole session.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task InvokeAsync_WhenCommandExceedsFrameMax_FailsOnlyThatCommandAndStaysReady()
{
await using PipePair pipePair = await PipePair.CreateAsync();
await using WorkerClient client = CreateClient(pipePair, maxMessageBytes: 4096);
await CompleteHandshakeAsync(client, pipePair);
WorkerCommand oversized = new()
{
Command = new MxCommand
{
Kind = MxCommandKind.Write,
Write = new WriteCommand
{
ServerHandle = 1,
ItemHandle = 2,
Value = new MxValue { StringValue = new string('x', 8192) },
},
},
};
WorkerClientException exception = await Assert.ThrowsAsync<WorkerClientException>(
async () => await client.InvokeAsync(oversized, TestTimeout, CancellationToken.None));
Assert.Equal(WorkerClientErrorCode.CommandTooLarge, exception.ErrorCode);
Assert.Equal(WorkerClientState.Ready, client.State);
// A subsequent normally-sized command still round-trips: the session was not faulted.
Task<WorkerCommandReply> nextInvoke = client.InvokeAsync(
CreateCommand(MxCommandKind.Ping),
TestTimeout,
CancellationToken.None);
WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
await pipePair.WorkerWriter.WriteAsync(
CreateCommandReplyEnvelope(commandEnvelope.CorrelationId, MxCommandKind.Ping));
WorkerCommandReply reply = await nextInvoke.WaitAsync(TestTimeout);
Assert.Equal(MxCommandKind.Ping, reply.Reply.Kind);
Assert.Equal(WorkerClientState.Ready, client.State);
}
/// <summary>Verifies that InvokeAsync ignores late replies and keeps the client ready.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
@@ -564,9 +610,13 @@ public sealed class WorkerClientTests
WorkerClientOptions? options = null,
GatewayMetrics? metrics = null,
WorkerProcessHandle? processHandle = null,
TimeProvider? timeProvider = null)
TimeProvider? timeProvider = null,
int maxMessageBytes = WorkerFrameProtocolOptions.DefaultMaxMessageBytes)
{
WorkerFrameProtocolOptions frameOptions = new(SessionId);
WorkerFrameProtocolOptions frameOptions = new(
SessionId,
GatewayContractInfo.WorkerProtocolVersion,
maxMessageBytes);
WorkerClientConnection connection = new(
SessionId,
Nonce,