diff --git a/docs/GatewayConfiguration.md b/docs/GatewayConfiguration.md index 3b1eaec..7dca44b 100644 --- a/docs/GatewayConfiguration.md +++ b/docs/GatewayConfiguration.md @@ -29,7 +29,7 @@ paths, timeouts, queue sizes, enum values, or protocol values are invalid. "ShutdownTimeoutSeconds": 10, "HeartbeatIntervalSeconds": 5, "HeartbeatGraceSeconds": 15, - "MaxMessageBytes": 16777216 + "MaxMessageBytes": 16842752 }, "Sessions": { "DefaultCommandTimeoutSeconds": 30, @@ -117,12 +117,16 @@ CWD (SEC-01). | `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`. | -| `MxGateway:Worker:MaxMessageBytes` | `16777216` | Maximum worker IPC frame payload size in bytes. The validator allows values from `1024` through `268435456`. | +| `MxGateway:Worker:MaxMessageBytes` | `16842752` | Maximum worker IPC frame payload size in bytes. The validator allows values from `1024` through `268435456`, and additionally requires this to be at least `MxGateway:Protocol:MaxGrpcMessageBytes` plus a 65536-byte (64 KiB) envelope-overhead reserve. The default is the 16 MiB public gRPC cap plus that reserve. The gateway conveys the negotiated value to the worker in the handshake (`GatewayHello.max_frame_bytes`), so the worker frames to the same limit rather than a hard-coded default. | `StartupProbeRetryAttempts`, `StartupProbeRetryDelayMilliseconds`, `PipeConnectAttemptTimeoutMilliseconds`, timeout values, heartbeat values, and `MaxMessageBytes` must be positive. `MaxMessageBytes` is intentionally bounded -to avoid accidental large allocations from malformed or oversized frames. +to avoid accidental large allocations from malformed or oversized frames. It +must also stay above `MaxGrpcMessageBytes` by the envelope-overhead reserve so a +maximally-sized accepted gRPC payload always fits one worker frame once wrapped +in a `WorkerEnvelope`; otherwise the oversized outbound write would fault the +whole session. Startup validation fails fast when the headroom is violated. ## Session Options @@ -256,7 +260,7 @@ dev→production hardening posture. | Option | Default | Description | |--------|---------|-------------| | `MxGateway:Protocol:WorkerProtocolVersion` | `1` | Worker IPC protocol version expected by the gateway and worker. This must match `GatewayContractInfo.WorkerProtocolVersion`. | -| `MxGateway:Protocol:MaxGrpcMessageBytes` | `16777216` | Public gRPC max send and receive message size in bytes. The same default is used by official clients. The validator allows values from `1024` through `268435456`. | +| `MxGateway:Protocol:MaxGrpcMessageBytes` | `16777216` | Public gRPC max send and receive message size in bytes. The same default is used by official clients. The validator allows values from `1024` through `268435456`, and requires `MxGateway:Worker:MaxMessageBytes` to exceed this by at least the 64 KiB worker-frame envelope reserve (see the Worker options). | The protocol option is exposed for diagnostics and explicit deployment configuration, not for compatibility negotiation. A mismatch fails validation diff --git a/docs/WorkerFrameProtocol.md b/docs/WorkerFrameProtocol.md index 8a43fbe..b74ff6e 100644 --- a/docs/WorkerFrameProtocol.md +++ b/docs/WorkerFrameProtocol.md @@ -17,8 +17,17 @@ payload_length bytes protobuf WorkerEnvelope ``` The reader rejects zero-length payloads and payloads larger than the configured -maximum before allocating the payload buffer. The default maximum is 16 MiB, -matching the gateway process design. +maximum before allocating the payload buffer. The default maximum is the 16 MiB +public gRPC cap plus a 64 KiB envelope-overhead reserve (16842752 bytes) so a +maximally-sized accepted gRPC payload always fits one worker frame once wrapped +in a `WorkerEnvelope`. + +The gateway is the source of truth for this maximum: it conveys the negotiated +value in the handshake as `GatewayHello.max_frame_bytes`, and the worker adopts +it as its `WorkerFrameProtocolOptions.MaxMessageBytes` instead of a hard-coded +default. A `max_frame_bytes` of 0 (an older gateway that never set the field) +means "use the worker's built-in default". This keeps both ends framing to the +same limit rather than depending on matched compile-time constants. ## Envelope Validation diff --git a/gateway.md b/gateway.md index 30267df..0c5fdf5 100644 --- a/gateway.md +++ b/gateway.md @@ -424,7 +424,10 @@ Optional diagnostics: - `Ping` - `GetSessionState` - `GetWorkerInfo` -- `DrainEvents` +- `DrainEvents` — diagnostic; `max_events` is bounded (the gateway rejects requests + above a public ceiling, and the worker caps each reply at its own per-reply limit, + treating `max_events = 0` as "the default cap") so one drain cannot pack an + unbounded, session-killing reply frame. - `ShutdownWorker` Do not compress MXAccess semantics into generic verbs too early. A command enum diff --git a/src/ZB.MOM.WW.MxGateway.Contracts/Generated/MxaccessWorker.cs b/src/ZB.MOM.WW.MxGateway.Contracts/Generated/MxaccessWorker.cs index 268af42..d8e1444 100644 --- a/src/ZB.MOM.WW.MxGateway.Contracts/Generated/MxaccessWorker.cs +++ b/src/ZB.MOM.WW.MxGateway.Contracts/Generated/MxaccessWorker.cs @@ -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 { } } + /// Field number for the "max_frame_bytes" field. + public const int MaxFrameBytesFieldNumber = 4; + private uint maxFrameBytes_; + /// + /// 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. + /// + [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; + } } } } diff --git a/src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_worker.proto b/src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_worker.proto index 06a22de..e50c4ff 100644 --- a/src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_worker.proto +++ b/src/ZB.MOM.WW.MxGateway.Contracts/Protos/mxaccess_worker.proto @@ -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 { diff --git a/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs b/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs index 34c50e7..d271063 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs @@ -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= 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); diff --git a/src/ZB.MOM.WW.MxGateway.Server/Configuration/WorkerOptions.cs b/src/ZB.MOM.WW.MxGateway.Server/Configuration/WorkerOptions.cs index 28d3386..baf68ea 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Configuration/WorkerOptions.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Configuration/WorkerOptions.cs @@ -33,6 +33,14 @@ public sealed class WorkerOptions /// The grace period in seconds after a heartbeat before considering the worker unresponsive. public int HeartbeatGraceSeconds { get; init; } = 15; - /// The maximum message size in bytes for IPC communication. - public int MaxMessageBytes { get; init; } = 16 * 1024 * 1024; + /// + /// The maximum worker-frame (pipe) message size in bytes. Must stay at least + /// above + /// 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 (GatewayHello.max_frame_bytes, IPC-02). Default is the 16 MB public gRPC cap + /// plus that reserve. + /// + public int MaxMessageBytes { get; init; } = + (16 * 1024 * 1024) + Workers.WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes; } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs index 6d6d8d2..22675f1 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs @@ -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, }; diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGrpcRequestValidator.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGrpcRequestValidator.cs index 5de7766..c6bc0e6 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGrpcRequestValidator.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGrpcRequestValidator.cs @@ -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; + /// Validates an open session request. /// The request to validate. 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) diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs index 64b4e42..b8e603a 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs @@ -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 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, }); } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClientErrorCode.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClientErrorCode.cs index a6a9542..8f32111 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClientErrorCode.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClientErrorCode.cs @@ -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, } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerFrameProtocolOptions.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerFrameProtocolOptions.cs index ad9d083..1b664d2 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerFrameProtocolOptions.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerFrameProtocolOptions.cs @@ -10,6 +10,15 @@ public sealed class WorkerFrameProtocolOptions /// Default maximum message size in bytes (16 MB). public const int DefaultMaxMessageBytes = 16 * 1024 * 1024; + /// + /// 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 WorkerEnvelope (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. + /// + public const int EnvelopeOverheadReserveBytes = 64 * 1024; + /// /// Initializes worker frame protocol options with a session ID. /// diff --git a/src/ZB.MOM.WW.MxGateway.Server/appsettings.json b/src/ZB.MOM.WW.MxGateway.Server/appsettings.json index cfe89ea..7623e47 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/appsettings.json +++ b/src/ZB.MOM.WW.MxGateway.Server/appsettings.json @@ -38,7 +38,7 @@ "ShutdownTimeoutSeconds": 10, "HeartbeatIntervalSeconds": 5, "HeartbeatGraceSeconds": 15, - "MaxMessageBytes": 16777216 + "MaxMessageBytes": 16842752 }, "Sessions": { "DefaultCommandTimeoutSeconds": 30, diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs index d2614a3..65c60f1 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs @@ -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, + }; + } + + /// + /// 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. + /// + [Fact] + public void Validate_Succeeds_WhenWorkerFrameMaxHasEnvelopeHeadroom() + { + ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, ValidOptions()); + Assert.True(result.Succeeded); + } + + /// + /// 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). + /// + [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")); + } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGrpcRequestValidatorTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGrpcRequestValidatorTests.cs new file mode 100644 index 0000000..52f9908 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGrpcRequestValidatorTests.cs @@ -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 }, + }, + }; + + /// + /// Verifies a DrainEvents request within the per-request ceiling passes validation, including the + /// max_events = 0 "worker default cap" sentinel (IPC-04). + /// + [Theory] + [InlineData(0u)] + [InlineData(1u)] + [InlineData(10_000u)] + public void ValidateInvoke_AllowsDrainEvents_WithinCeiling(uint maxEvents) + { + MxAccessGrpcRequestValidator validator = new(); + validator.ValidateInvoke(DrainRequest(maxEvents)); + } + + /// + /// 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). + /// + [Fact] + public void ValidateInvoke_RejectsDrainEvents_AboveCeiling() + { + MxAccessGrpcRequestValidator validator = new(); + RpcException exception = Assert.Throws(() => validator.ValidateInvoke(DrainRequest(10_001))); + Assert.Equal(StatusCode.InvalidArgument, exception.StatusCode); + Assert.Contains("max_events", exception.Status.Detail, StringComparison.Ordinal); + } +} diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs index 53411bb..b3d38a5 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs @@ -56,6 +56,52 @@ public sealed class WorkerClientTests Assert.Equal(MxCommandKind.Ping, reply.Reply.Kind); } + /// + /// Verifies that a command whose serialized envelope exceeds the negotiated worker-frame maximum + /// fails only that command with 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. + /// + /// A task that represents the asynchronous operation. + [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( + 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 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); + } + /// Verifies that InvokeAsync ignores late replies and keeps the client ready. /// A task that represents the asynchronous operation. [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,