From 5efd33ec4c4e739ed617a0d224e551086867bf31 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Fri, 13 Nov 2026 16:02:03 +0100 Subject: [PATCH 1/3] fix(ingest): keep three request costs bounded per request - A session replay chunk is inflated on the blocking pool, like an OTLP body, so a chunk that inflates to the decoded ceiling does not hold the async workers. - A Vercel audit delivery is refused past 10,000 events. Every event becomes a row of fixed columns, so the byte ceiling alone did not bound what a delivery stores. - A credential longer than any key Maple issues is treated as no credential, before it is hashed, looked up or cached as unknown. --- apps/ingest/src/main.rs | 80 +++++++++++++++++++++++++++++++++++------ 1 file changed, 70 insertions(+), 10 deletions(-) diff --git a/apps/ingest/src/main.rs b/apps/ingest/src/main.rs index 21e390e8ae..e1bd8c224b 100644 --- a/apps/ingest/src/main.rs +++ b/apps/ingest/src/main.rs @@ -4198,6 +4198,11 @@ const AUDIT_TRAIL_GCP_ORG_RPS: u32 = 100; /// events is kilobytes; the telemetry ceiling is for bodies that are metered. const AUDIT_TRAIL_MAX_DECODED_BYTES: usize = 4 * 1024 * 1024; +/// The most events one Vercel audit delivery may carry. Every event becomes a +/// row of fixed columns however short it is, so the byte ceiling alone does not +/// bound what a delivery stores or holds in memory. +const AUDIT_TRAIL_MAX_EVENTS: usize = 10_000; + /// Refuses an org's audit delivery past `rps`. The rows are not metered, so /// this is what bounds how many a sender can store. async fn audit_trail_rate_gate( @@ -4372,6 +4377,9 @@ async fn decode_vercel_audit_drain( let parsed = vercel_logs::parse(&payload).map_err(|e| { ApiError::bad_request(format!("Invalid Vercel Audit Log drain payload: {e}")) })?; + if parsed.entries.len() > AUDIT_TRAIL_MAX_EVENTS { + return Err(too_large()); + } let now_ms = audit_trail::now_ms(); let rows: Vec> = parsed .entries @@ -5515,14 +5523,26 @@ async fn handle_replay_blob_inner( GunzipError::Invalid(e) => replay_gunzip_rejection(headers, &body, &e), GunzipError::TooLarge => ApiError::payload_too_large("decompressed replay chunk too large"), }; - let (events_json, byte_size) = if state.replay_blob_store.is_some() { - let byte_size = gunzip_len_bounded(&body, max_decoded).map_err(gunzip_rejection)?; - (None, byte_size) - } else { - let events_json = gunzip_replay_text(&body, max_decoded).map_err(gunzip_rejection)?; - let byte_size = events_json.len() as u64; - (Some(events_json), byte_size) - }; + // Inflating is CPU work bounded only by that ceiling, so it runs on the + // blocking pool: on a one-core host the async workers are a single thread. + let stored_as_blob = state.replay_blob_store.is_some(); + let compressed = body.clone(); + let (events_json, byte_size) = tokio::task::spawn_blocking(move || { + if stored_as_blob { + gunzip_len_bounded(&compressed, max_decoded).map(|byte_size| (None, byte_size)) + } else { + gunzip_replay_text(&compressed, max_decoded).map(|events_json| { + let byte_size = events_json.len() as u64; + (Some(events_json), byte_size) + }) + } + }) + .await + .map_err(|join_error| { + error!(error = %join_error, "Replay chunk decode task failed"); + ApiError::service_unavailable("Ingest gateway failed to decode the request") + })? + .map_err(gunzip_rejection)?; // Accept the chunk that crosses the ceiling so the recording truncates on a // chunk boundary; `is_exhausted` rejects everything after it. @@ -7484,6 +7504,10 @@ fn json_value_to_attribute(key: &str, value: &JsonValue) -> Option { Some(string_attribute(key, &string_value)) } +/// Longer than any key Maple issues. A longer value is not a key, and is +/// refused here so it is never hashed, looked up or held in the unknown-key cache. +const MAX_INGEST_KEY_LEN: usize = 256; + fn extract_ingest_key(headers: &HeaderMap) -> Option { if let Some(value) = headers .get(AUTHORIZATION) @@ -7492,7 +7516,7 @@ fn extract_ingest_key(headers: &HeaderMap) -> Option { if value.len() > 7 && value[..7].eq_ignore_ascii_case("Bearer ") { let token = value[7..].trim(); if !token.is_empty() { - return Some(token.to_owned()); + return (token.len() <= MAX_INGEST_KEY_LEN).then(|| token.to_owned()); } } } @@ -7501,7 +7525,7 @@ fn extract_ingest_key(headers: &HeaderMap) -> Option { .get("x-maple-ingest-key") .and_then(|value| value.to_str().ok()) .map(str::trim) - .filter(|value| !value.is_empty()) + .filter(|value| !value.is_empty() && value.len() <= MAX_INGEST_KEY_LEN) .map(str::to_owned) } @@ -11535,6 +11559,22 @@ mod tests { assert!(!is_sentinel_token("maple_pk_123")); } + #[test] + fn extract_ingest_key_ignores_a_value_longer_than_any_key() { + let oversized = format!("maple_pk_{}", "a".repeat(MAX_INGEST_KEY_LEN)); + let mut bearer = HeaderMap::new(); + bearer.insert(AUTHORIZATION, format!("Bearer {oversized}").parse().unwrap()); + assert_eq!(extract_ingest_key(&bearer), None); + let mut named = HeaderMap::new(); + named.insert("x-maple-ingest-key", oversized.parse().unwrap()); + assert_eq!(extract_ingest_key(&named), None); + + let longest = "k".repeat(MAX_INGEST_KEY_LEN); + let mut accepted = HeaderMap::new(); + accepted.insert("x-maple-ingest-key", longest.parse().unwrap()); + assert_eq!(extract_ingest_key(&accepted), Some(longest)); + } + #[test] fn extract_ingest_key_returns_sentinel_literal_unchanged() { let mut headers = HeaderMap::new(); @@ -13606,6 +13646,26 @@ mod tests { drop(std::fs::remove_dir_all(&queue_dir)); } + // Every event is a stored row whatever its size, so a small body of many + // events is refused on its count. + #[tokio::test] + async fn vercel_audit_drain_refuses_a_delivery_of_more_events_than_any_drain_sends() { + let queue_dir = unique_main_test_dir("vercel-audit-trail-count"); + let (state, _store, mut imports) = + audit_trail_test_state("maple_sk_test_count", "org_count", queue_dir.clone()).await; + let headers = vercel_drain_headers("maple_sk_test_count", AUDIT_SOURCE_ID); + let body = Bytes::from("{\"action\":\"a\"}\n".repeat(AUDIT_TRAIL_MAX_EVENTS + 1)); + assert!(body.len() < AUDIT_TRAIL_MAX_DECODED_BYTES); + + let refused = handle_vercel_audit_trail_inner(&state, &headers, &body) + .await + .expect_err("a delivery past the event limit must be refused"); + + assert_eq!(refused.status, StatusCode::PAYLOAD_TOO_LARGE); + assert!(nothing_reaches_the_audit_trail(&mut imports).await); + drop(std::fs::remove_dir_all(&queue_dir)); + } + // Audit rows are not metered, so the rate is what bounds a sender. #[tokio::test] async fn an_orgs_audit_deliveries_are_refused_past_their_rate() { From e574a7d6b238e2225d902cd2eb7e900d010d5444 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Fri, 13 Nov 2026 16:09:22 +0100 Subject: [PATCH 2/3] fix(ingest): replay decode reserves from the shared decode memory On the blocking pool many chunks can inflate at once, so each reserves its body and declared size first and is shed with the retryable busy answer when that does not fit. An inline chunk keeps its text reserved until it is stored. --- apps/ingest/src/main.rs | 54 +++++++++++++++++++++++++++++++++++------ 1 file changed, 46 insertions(+), 8 deletions(-) diff --git a/apps/ingest/src/main.rs b/apps/ingest/src/main.rs index e1bd8c224b..eb4b8403b0 100644 --- a/apps/ingest/src/main.rs +++ b/apps/ingest/src/main.rs @@ -5525,24 +5525,48 @@ async fn handle_replay_blob_inner( }; // Inflating is CPU work bounded only by that ceiling, so it runs on the // blocking pool: on a one-core host the async workers are a single thread. + // It reserves what it inflates from the shared decode memory, like an OTLP + // body, so a burst of chunks is shed and not inflated all at once. let stored_as_blob = state.replay_blob_store.is_some(); let compressed = body.clone(); - let (events_json, byte_size) = tokio::task::spawn_blocking(move || { + let mut reservation = state.decode_memory.reservation(); + let decoded = tokio::task::spawn_blocking(move || { + let ceiling = u64::try_from(max_decoded).unwrap_or(u64::MAX); + let declared = decode_budget::gzip_declared_len(&compressed) + .map_or(0, |declared| declared.min(ceiling)); + if !reservation.try_add(compressed.len() as u64 + declared) { + return Err(None); + } if stored_as_blob { - gunzip_len_bounded(&compressed, max_decoded).map(|byte_size| (None, byte_size)) + // Nothing of a blob chunk outlives the count. + gunzip_len_bounded(&compressed, max_decoded) + .map(|byte_size| (None, byte_size, None)) + .map_err(Some) } else { - gunzip_replay_text(&compressed, max_decoded).map(|events_json| { - let byte_size = events_json.len() as u64; - (Some(events_json), byte_size) - }) + gunzip_replay_text(&compressed, max_decoded) + .map(|events_json| { + let byte_size = events_json.len() as u64; + reservation.shrink_to(byte_size); + (Some(events_json), byte_size, Some(reservation)) + }) + .map_err(Some) } }) .await .map_err(|join_error| { error!(error = %join_error, "Replay chunk decode task failed"); ApiError::service_unavailable("Ingest gateway failed to decode the request") - })? - .map_err(gunzip_rejection)?; + })?; + // An inline chunk's text stays reserved until this handler has stored it. + let (events_json, byte_size, _inline_text_reservation) = + decoded.map_err(|rejection| match rejection { + Some(error) => gunzip_rejection(error), + None => ApiError::tagged( + StatusCode::SERVICE_UNAVAILABLE, + &INGEST_DECODE_BUSY, + "Ingest gateway is busy decoding other requests; retry shortly", + ), + })?; // Accept the chunk that crosses the ceiling so the recording truncates on a // chunk boundary; `is_exhausted` rejects everything after it. @@ -13876,6 +13900,20 @@ mod tests { ) .await .expect("a normal chunk is still accepted"); + + // A chunk the shared decode memory cannot hold right now is shed, not inflated. + let mut others = state.decode_memory.reservation(); + assert!(others.try_add(state.decode_memory.total_bytes())); + let shed = handle_replay_blob_inner( + &state, + &replay_blob_headers(raw_key, "sess_bomb", 2), + gzip_bytes(r#"[{"type":2,"timestamp":2}]"#), + ) + .await + .expect_err("a chunk must be shed while decode memory is full"); + assert_eq!(shed.status, StatusCode::SERVICE_UNAVAILABLE); + drop(others); + assert_eq!(state.decode_memory.reserved_bytes(), 0); assert!(rx.recv().await.is_some(), "the normal chunk is stored"); drop(std::fs::remove_dir_all(&queue_dir)); From 58bcbb1a9aa472d158f6b0f1febfb18f3fe17e6c Mon Sep 17 00:00:00 2001 From: Makisuo Date: Fri, 13 Nov 2026 16:14:55 +0100 Subject: [PATCH 3/3] fix(ingest): a replay chunk inflates no further than the size it reserved The reservation is sized from the gzip trailer, which the sender writes. The inflate now stops at that declared size, so a trailer that understates the stream is refused and cannot get more decoded than was reserved. --- apps/ingest/src/main.rs | 25 ++++++++++++++++++++----- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/apps/ingest/src/main.rs b/apps/ingest/src/main.rs index eb4b8403b0..07744e5675 100644 --- a/apps/ingest/src/main.rs +++ b/apps/ingest/src/main.rs @@ -5531,19 +5531,21 @@ async fn handle_replay_blob_inner( let compressed = body.clone(); let mut reservation = state.decode_memory.reservation(); let decoded = tokio::task::spawn_blocking(move || { - let ceiling = u64::try_from(max_decoded).unwrap_or(u64::MAX); + // The size the chunk declares is what gets reserved, so it is also where + // the inflate stops: a trailer that understates the stream buys nothing. let declared = decode_budget::gzip_declared_len(&compressed) - .map_or(0, |declared| declared.min(ceiling)); - if !reservation.try_add(compressed.len() as u64 + declared) { + .map_or(0, |declared| usize::try_from(declared).unwrap_or(usize::MAX)) + .min(max_decoded); + if !reservation.try_add((compressed.len() + declared) as u64) { return Err(None); } if stored_as_blob { // Nothing of a blob chunk outlives the count. - gunzip_len_bounded(&compressed, max_decoded) + gunzip_len_bounded(&compressed, declared) .map(|byte_size| (None, byte_size, None)) .map_err(Some) } else { - gunzip_replay_text(&compressed, max_decoded) + gunzip_replay_text(&compressed, declared) .map(|events_json| { let byte_size = events_json.len() as u64; reservation.shrink_to(byte_size); @@ -13901,6 +13903,19 @@ mod tests { .await .expect("a normal chunk is still accepted"); + // A trailer that understates the stream does not get more inflated than it reserved. + let mut understated = gzip_bytes(&"a".repeat(64 * 1024)).to_vec(); + let trailer = understated.len() - 4; + understated[trailer..].copy_from_slice(&8_u32.to_le_bytes()); + let error = handle_replay_blob_inner( + &state, + &replay_blob_headers(raw_key, "sess_bomb", 3), + Bytes::from(understated), + ) + .await + .expect_err("a chunk inflating past its declared size must be refused"); + assert_eq!(error.status, StatusCode::PAYLOAD_TOO_LARGE); + // A chunk the shared decode memory cannot hold right now is shed, not inflated. let mut others = state.decode_memory.reservation(); assert!(others.try_add(state.decode_memory.total_bytes()));