environmentd: make webhook max request size configurable (#37277)

### Motivation

The webhook source request body limit was a compile-time constant (5
MiB) applied to all HTTP routes, so operators could not adjust it
without a rebuild.

### Description

Adds the `webhook_max_request_size_bytes` dyncfg (default 5 MiB,
preserving prior behavior). It is read from the live shared persist
`ConfigSet`, so there is no coordinator round-trip on the webhook hot
path. The webhook route disables the global `DefaultBodyLimit` and the
handler enforces the cap per request on the decompressed body via
`to_bytes`: size-exceeded returns 413, other body-read errors return
500. Enforcement is scoped to the webhook route only; SQL and other HTTP
routes keep the static 5 MiB limit.

### Verification

New integration test `webhook_max_request_size` in
`src/environmentd/tests/server.rs`: the 5 MiB default rejects a 6 MiB
body, raising the dyncfg accepts it, and lowering it below the default
re-rejects.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: antiguru

misc/python/materialize/mzcompose/__init__.py

@@ -679,6 +679,7 @@ def get_default_system_parameters(
     "enable_mcp_developer_query_tool",
     "mcp_max_response_size",
     "user_id_pool_batch_size",
+    "webhook_max_request_size_bytes",
 ]
 
 

misc/python/materialize/parallel_workload/action.py

@@ -1655,6 +1655,12 @@ def __init__(
             BOOLEAN_FLAG_VALUES
         )
         self.flags_with_values["enable_upsert_paged_spill"] = BOOLEAN_FLAG_VALUES
+        self.flags_with_values["webhook_max_request_size_bytes"] = [
+            # 1 MiB, 5 MiB (default), 10 MiB
+            "1048576",
+            "5242880",
+            "10485760",
+        ]
 
         # If you are adding a new config flag in Materialize, consider using it
         # here instead of just marking it as uninteresting to silence the

src/adapter-types/src/dyncfgs.rs

@@ -221,6 +221,17 @@ pub const MCP_MAX_RESPONSE_SIZE: Config<usize> = Config::new(
     "Maximum size in bytes of MCP tool response content. Responses exceeding this limit are rejected with an error telling the agent to narrow its query.",
 );
 
+/// Maximum size (in bytes) of a webhook request body, measured after
+/// decompression. Requests whose body exceeds this limit are rejected with
+/// HTTP 413. Applies only to the webhook route; other HTTP routes use a
+/// separate static limit.
+pub const WEBHOOK_MAX_REQUEST_SIZE_BYTES: Config<usize> = Config::new(
+    "webhook_max_request_size_bytes",
+    // Matches `MAX_REQUEST_SIZE`, the static limit the other environmentd HTTP routes use.
+    5 * 1024 * 1024,
+    "The maximum size in bytes of a webhook request body, measured after decompression.",
+);
+
 /// Number of user IDs to pre-allocate in a batch. Pre-allocating IDs avoids
 /// a persist write + oracle call per DDL statement.
 pub const USER_ID_POOL_BATCH_SIZE: Config<u32> = Config::new(
@@ -319,6 +330,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&ENABLE_MCP_DEVELOPER_QUERY_TOOL)
         .add(&ENABLE_PUBLIC_METRICS_ENDPOINT)
         .add(&MCP_MAX_RESPONSE_SIZE)
+        .add(&WEBHOOK_MAX_REQUEST_SIZE_BYTES)
         .add(&USER_ID_POOL_BATCH_SIZE)
         .add(&CONSOLE_OIDC_CLIENT_ID)
         .add(&CONSOLE_OIDC_SCOPES)

src/environmentd/src/environmentd/main.rs

@@ -1022,6 +1022,7 @@ fn run(mut args: Args) -> Result<(), anyhow::Error> {
     });
 
     let persist_clients = Arc::new(persist_clients);
+    let system_dyncfgs = Arc::clone(&persist_clients.cfg().configs);
     let connection_context = ConnectionContext::from_cli_args(
         args.environment_id.to_string(),
         &args.tracing.startup_log_filter,
@@ -1101,6 +1102,7 @@ fn run(mut args: Args) -> Result<(), anyhow::Error> {
                 controller,
                 secrets_controller,
                 cloud_resource_controller,
+                system_dyncfgs,
                 // Storage options.
                 storage_usage_collection_interval: args.storage_usage_collection_interval_sec,
                 storage_usage_retention_period: args.storage_usage_retention_period,

src/environmentd/src/http.rs

@@ -80,6 +80,7 @@ use mz_auth::Authenticated;
 use mz_auth::password::Password;
 use mz_authenticator::Authenticator;
 use mz_controller::ReplicaHttpLocator;
+use mz_dyncfg::ConfigSet;
 use mz_frontegg_auth::Error as FronteggError;
 use mz_http_util::DynamicFilterTarget;
 use mz_ore::cast::u64_to_usize;
@@ -174,6 +175,7 @@ pub struct HttpConfig {
     pub http_host_name: Option<String>,
     pub frontegg_oauth_issuer_url: Option<String>,
     pub concurrent_webhook_req: Arc<tokio::sync::Semaphore>,
+    pub dyncfgs: Arc<ConfigSet>,
     pub metrics: Metrics,
     pub metrics_registry: MetricsRegistry,
     pub mcp_metrics: mcp_metrics::McpMetrics,
@@ -206,6 +208,7 @@ pub struct WsState {
 pub struct WebhookState {
     adapter_client_rx: Delayed<mz_adapter::Client>,
     webhook_cache: WebhookAppenderCache,
+    dyncfgs: Arc<ConfigSet>,
 }
 
 #[derive(Clone, Debug)]
@@ -233,6 +236,7 @@ impl HttpServer {
             http_host_name,
             frontegg_oauth_issuer_url,
             concurrent_webhook_req,
+            dyncfgs,
             metrics,
             metrics_registry,
             mcp_metrics,
@@ -346,6 +350,7 @@ impl HttpServer {
                 .with_state(WebhookState {
                     adapter_client_rx: adapter_client_rx.clone(),
                     webhook_cache,
+                    dyncfgs,
                 })
                 .layer(
                     tower_http::decompression::RequestDecompressionLayer::new()
@@ -354,6 +359,11 @@ impl HttpServer {
                         .br(true)
                         .zstd(true),
                 )
+                // The webhook handler enforces WEBHOOK_MAX_REQUEST_SIZE_BYTES
+                // itself via to_bytes on a raw Body. This disable is defense-in-depth:
+                // it only matters if the handler ever returns to a Bytes extractor,
+                // which would otherwise inherit the global 5 MiB limit.
+                .layer(DefaultBodyLimit::disable())
                 .layer(
                     CorsLayer::new()
                         .allow_methods(Method::POST)

src/environmentd/src/http/webhook.rs

@@ -21,10 +21,12 @@ use mz_repr::{Datum, Diff, Row, RowPacker, SqlScalarType};
 use mz_sql::plan::{WebhookBodyFormat, WebhookHeaderFilters, WebhookHeaders};
 use mz_storage_types::controller::StorageError;
 
+use axum::body::Body;
 use axum::extract::{Path, State};
 use axum::response::IntoResponse;
 use bytes::Bytes;
 use http::StatusCode;
+use mz_adapter_types::dyncfgs::WEBHOOK_MAX_REQUEST_SIZE_BYTES;
 use thiserror::Error;
 
 use crate::http::WebhookState;
@@ -33,11 +35,32 @@ pub async fn handle_webhook(
     State(WebhookState {
         adapter_client_rx,
         webhook_cache,
+        dyncfgs,
     }): State<WebhookState>,
     Path((database, schema, name)): Path<(String, String, String)>,
     headers: http::HeaderMap,
-    body: Bytes,
+    body: Body,
 ) -> impl IntoResponse {
+    let max_request_size = WEBHOOK_MAX_REQUEST_SIZE_BYTES.get(&dyncfgs);
+    let body = axum::body::to_bytes(body, max_request_size)
+        .await
+        .map_err(|err| {
+            use std::error::Error;
+            // axum::Error wraps the underlying cause as its source. If that source is a
+            // LengthLimitError the body exceeded the configured limit (HTTP 413). Any other
+            // cause (TCP reset, decompression failure, etc.) is an internal read error (HTTP 500)
+            // and must not be reported as a size-limit violation.
+            if err
+                .source()
+                .is_some_and(|s| s.is::<http_body_util::LengthLimitError>())
+            {
+                WebhookError::BodyTooLarge {
+                    max_bytes: max_request_size,
+                }
+            } else {
+                WebhookError::Internal(anyhow::anyhow!(err))
+            }
+        })?;
     let adapter_client = adapter_client_rx.clone().await.expect("sender not dropped");
     // Collect headers into a map, while converting them into strings.
     let mut headers_s = BTreeMap::new();
@@ -332,6 +355,8 @@ pub enum WebhookError {
     Unavailable,
     #[error("internal storage failure! {0:?}")]
     InternalStorageError(StorageError),
+    #[error("request body exceeds the maximum allowed size of {max_bytes} bytes")]
+    BodyTooLarge { max_bytes: usize },
     #[error("internal failure! {0:?}")]
     Internal(#[from] anyhow::Error),
 }
@@ -387,6 +412,9 @@ impl IntoResponse for WebhookError {
             e @ WebhookError::InvalidHeaders(_) => {
                 (StatusCode::UNAUTHORIZED, e.to_string()).into_response()
             }
+            e @ WebhookError::BodyTooLarge { .. } => {
+                (StatusCode::PAYLOAD_TOO_LARGE, e.to_string()).into_response()
+            }
             e @ WebhookError::Unavailable => {
                 (StatusCode::SERVICE_UNAVAILABLE, e.to_string()).into_response()
             }

src/environmentd/src/lib.rs

@@ -43,6 +43,7 @@ use mz_catalog::config::ClusterReplicaSizeMap;
 use mz_catalog::durable::BootstrapArgs;
 use mz_cloud_resources::CloudResourceController;
 use mz_controller::ControllerConfig;
+use mz_dyncfg::ConfigSet;
 use mz_frontegg_auth::Authenticator as FronteggAuthenticator;
 use mz_license_keys::ValidatedLicenseKey;
 use mz_ore::future::OreFutureExt;
@@ -137,6 +138,8 @@ pub struct Config {
     pub secrets_controller: Arc<dyn SecretsController>,
     /// VpcEndpoint controller configuration.
     pub cloud_resource_controller: Option<Arc<dyn CloudResourceController>>,
+    /// The process-wide live system dyncfg set.
+    pub system_dyncfgs: Arc<ConfigSet>,
 
     // === Storage options. ===
     /// The interval at which to collect storage usage information.
@@ -412,6 +415,7 @@ impl Listeners {
                 allowed_origin: config.cors_allowed_origin.clone(),
                 allowed_origin_list: config.cors_allowed_origin_list.clone(),
                 concurrent_webhook_req: webhook_concurrency_limit.semaphore(),
+                dyncfgs: Arc::clone(&config.system_dyncfgs),
                 metrics: metrics.clone(),
                 metrics_registry: metrics_registry.clone(),
                 mcp_metrics: mcp_metrics.clone(),

src/environmentd/src/test_util.rs

@@ -806,6 +806,7 @@ impl Listeners {
         let persist_clients =
             PersistClientCache::new(persist_cfg, &metrics_registry, |_, _| persist_pubsub_client);
         let persist_clients = Arc::new(persist_clients);
+        let system_dyncfgs = Arc::clone(&persist_clients.cfg().configs);
 
         let secrets_controller = Arc::clone(&orchestrator);
         let connection_context = ConnectionContext::for_tests(orchestrator.reader());
@@ -884,6 +885,7 @@ impl Listeners {
                 },
                 secrets_controller,
                 cloud_resource_controller: None,
+                system_dyncfgs,
                 tls: config.tls,
                 frontegg: config.frontegg,
                 frontegg_oauth_issuer_url: None,

src/environmentd/tests/server.rs

@@ -3911,6 +3911,93 @@ fn webhook_too_large_request() {
     assert_eq!(payload_too_large, 1.0);
 }
 
+#[mz_ore::test]
+#[cfg_attr(miri, ignore)] // too slow
+#[allow(clippy::disallowed_methods)]
+fn webhook_max_request_size() {
+    let server = test_util::TestHarness::default()
+        .unsafe_mode()
+        .start_blocking();
+
+    let mut mz_client = server
+        .pg_config_internal()
+        .user(&SYSTEM_USER.name)
+        .connect(postgres::NoTls)
+        .unwrap();
+
+    let mut client = server.connect(postgres::NoTls).unwrap();
+
+    client
+        .execute(
+            "CREATE CLUSTER webhook_cluster (SIZE 'scale=1,workers=1');",
+            &[],
+        )
+        .expect("failed to create cluster");
+    client
+        .execute(
+            "CREATE SOURCE webhook_bytes IN CLUSTER webhook_cluster \
+             FROM WEBHOOK BODY FORMAT BYTES",
+            &[],
+        )
+        .expect("failed to create source");
+
+    let http_client = reqwest::Client::new();
+    let webhook_url = format!(
+        "http://{}/api/webhook/materialize/public/webhook_bytes",
+        server.http_local_addr(),
+    );
+
+    // Helper: POST a body of `len` bytes, return the HTTP status.
+    let post = |len: usize| {
+        let body = vec![b'a'; len];
+        server.runtime().block_on(async {
+            http_client
+                .post(&webhook_url)
+                .body(body)
+                .send()
+                .await
+                .expect("request failed")
+                .status()
+        })
+    };
+
+    // Default is 5 MiB: a 6 MiB body must be rejected with 413.
+    assert_eq!(post(6 * 1024 * 1024).as_u16(), 413);
+
+    // Raise the limit to 10 MiB and confirm the 6 MiB body is now accepted.
+    mz_client
+        .batch_execute("ALTER SYSTEM SET webhook_max_request_size_bytes = 10485760")
+        .unwrap();
+    // The dyncfg propagates to the shared persist ConfigSet asynchronously,
+    // so retry briefly until the new limit takes effect.
+    Retry::default()
+        .max_duration(std::time::Duration::from_secs(30))
+        .retry(|_| {
+            if post(6 * 1024 * 1024).is_success() {
+                Ok(())
+            } else {
+                Err(())
+            }
+        })
+        .expect("6 MiB body accepted after raising the limit to 10 MiB");
+
+    // Lower the limit below the default and confirm enforcement is live: a
+    // body that would have been accepted at the default is now rejected.
+    mz_client
+        .batch_execute("ALTER SYSTEM SET webhook_max_request_size_bytes = 1024")
+        .unwrap();
+    Retry::default()
+        .max_duration(std::time::Duration::from_secs(30))
+        .retry(|_| {
+            if post(2048).as_u16() == 413 {
+                Ok(())
+            } else {
+                Err(())
+            }
+        })
+        .expect("2 KiB body rejected after lowering the limit to 1 KiB");
+}
+
 #[mz_ore::test]
 #[cfg_attr(miri, ignore)] // too slow
 #[allow(clippy::disallowed_methods)]

src/sqllogictest/src/runner.rs

@@ -1193,6 +1193,7 @@ impl<'a> RunnerInner<'a> {
             persist_clients: Arc::clone(&persist_clients),
             metrics: Arc::new(mz_catalog::durable::Metrics::new(&MetricsRegistry::new())),
         };
+        let system_dyncfgs = Arc::clone(&persist_clients.cfg().configs);
         let server_config = mz_environmentd::Config {
             catalog_config,
             timestamp_oracle_url: Some(timestamp_oracle_url),
@@ -1227,6 +1228,7 @@ impl<'a> RunnerInner<'a> {
             },
             secrets_controller,
             cloud_resource_controller: None,
+            system_dyncfgs,
             tls: None,
             frontegg: None,
             frontegg_oauth_issuer_url: None,