[BUG](log): Cap the log client's encode size at the log server's decode limit

rust/log-service/src/lib.rs

@@ -5493,6 +5493,27 @@ mod tests {
         assert_eq!(100, config.grpc.max_concurrent_streams);
     }
 
+    /// A client must never be able to send a message this server will refuse to decode.
+    ///
+    /// The write path is asymmetric: the frontend encodes a whole batch of log records into one
+    /// `PushLogsRequest` and this server decodes it. If the client's encode ceiling were the
+    /// larger of the two, an oversized batch would leave the frontend successfully and die at
+    /// decode here, far from the code that built it. Keeping the client's ceiling at or below the
+    /// server's makes the frontend reject it locally instead, with the offending size in the error.
+    #[test]
+    fn log_client_encode_limit_fits_server_decode_limit() {
+        let server = LogServerConfig::default();
+        let server_decode_limit = server
+            .max_decoding_message_size
+            .unwrap_or(server.grpc.max_decoding_message_size);
+        let client_encode_limit = GrpcLogConfig::default().max_encoding_message_size;
+        assert!(
+            client_encode_limit <= server_decode_limit,
+            "log client would encode up to {client_encode_limit} bytes but the log server \
+             decodes at most {server_decode_limit} bytes",
+        );
+    }
+
     #[test]
     fn opentelemetry_config_defaults() {
         let config = OpenTelemetryConfig {

rust/log/src/config.rs

@@ -7,6 +7,11 @@ pub struct GrpcLogConfig {
     pub connect_timeout_ms: u64,
     #[serde(default = "GrpcLogConfig::default_request_timeout_ms")]
     pub request_timeout_ms: u64,
+    /// Largest `PushLogsRequest` this client will put on the wire.
+    ///
+    /// Must stay at or below the log service's decode limit, otherwise an oversized batch leaves
+    /// the client fine and is rejected on arrival, with nothing in the error naming the request
+    /// that caused it.
     #[serde(default = "GrpcLogConfig::default_max_encoding_message_size")]
     pub max_encoding_message_size: usize,
     #[serde(default = "GrpcLogConfig::default_max_decoding_message_size")]
@@ -28,6 +33,7 @@ impl GrpcLogConfig {
         5000
     }
 
+    /// Matches the log service's default decode limit.
     fn default_max_encoding_message_size() -> usize {
         32_000_000
     }

rust/log/src/grpc_log.rs

@@ -341,7 +341,8 @@ impl Configurable<(GrpcLogConfig, System)> for GrpcLog {
             my_config.connect_timeout_ms,
             my_config.request_timeout_ms,
             my_config.port,
-            ClientOptions::new(Some(my_config.max_decoding_message_size)),
+            ClientOptions::new(Some(my_config.max_decoding_message_size))
+                .with_max_encoding_message_size(Some(my_config.max_encoding_message_size)),
         );
         let client_manager_handle = system.start_component(client_manager);