From 6a7e93419cd1c5ab28b5e0b467ac3c539bbf5314 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 20 Aug 2026 14:34:31 -0400 Subject: [PATCH 1/2] test(eventhubs): cover the send_event link size check Issue #5101 reports that `send_event` accepts an event larger than the sender link allows. The AMQP library splits the oversized payload across transfer frames instead of refusing it, so the event reaches the partition and the caller gets no size protection. These tests pin the refusal before any fix exists. Four offline unit tests drive `ProducerClient::check_message_size`. They pin the refusal above the link maximum, the inclusive boundary at the maximum, the unchanged behavior below it, and the skip when the link reports no maximum. AMQP 1.0 section 2.7.3 gives an unset maximum the meaning "no limit", so the send path must not invent one. One live test sends a 2 MiB event through both `send_event` and `send_message`, asserts the error kind, then sends a small event on the same client to show the link is still up. The tests fail to compile against the unchanged source, because neither `check_message_size` nor `ErrorKind::MessageSizeExceeded` exists yet. --- .../src/producer/mod.rs | 56 ++++++++++++++++- .../tests/eventhubs_producer.rs | 62 +++++++++++++++++++ 2 files changed, 117 insertions(+), 1 deletion(-) diff --git a/sdk/eventhubs/azure_messaging_eventhubs/src/producer/mod.rs b/sdk/eventhubs/azure_messaging_eventhubs/src/producer/mod.rs index 3c6432057d3..d9539fbb5ae 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/src/producer/mod.rs +++ b/sdk/eventhubs/azure_messaging_eventhubs/src/producer/mod.rs @@ -840,7 +840,9 @@ pub mod builders { #[cfg(test)] mod tests { use crate::common::tests::force_errors; - use crate::{models::EventData, EventDataBatchOptions, ProducerClient, Result}; + use crate::{ + error::ErrorKind, models::EventData, EventDataBatchOptions, ProducerClient, Result, + }; use azure_core::time::Duration; use azure_core_amqp::{error::AmqpErrorKind, AmqpTransport}; use azure_core_test::{recorded, TestContext}; @@ -1285,4 +1287,56 @@ mod tests { Ok(()) } + + const LINK_MAX_SIZE: u64 = 1_048_576; + + // An event larger than the sender link allows must be refused before the + // send, and the error must name both sizes. + #[test] + fn message_size_above_the_link_maximum_is_rejected() { + let error = ProducerClient::check_message_size(LINK_MAX_SIZE + 1, Some(LINK_MAX_SIZE)) + .expect_err("a message above the link maximum must be refused"); + assert!( + matches!( + error.kind, + ErrorKind::MessageSizeExceeded { + requested: 1_048_577, + max_allowed: LINK_MAX_SIZE, + } + ), + "the caller must be able to match on the kind, got: {error:?}" + ); + let message = error.to_string(); + assert!( + message.contains("1048577") && message.contains("1048576"), + "the error must name the requested and the allowed size, got: {message}" + ); + } + + // The boundary is inclusive, so the check must use `>` and not `>=`. + #[test] + fn message_size_equal_to_the_link_maximum_is_allowed() { + ProducerClient::check_message_size(LINK_MAX_SIZE, Some(LINK_MAX_SIZE)) + .expect("a message exactly at the link maximum is still sent"); + } + + // Every message the link can carry must go, as it does today. + #[test] + fn message_size_below_the_link_maximum_is_allowed() { + ProducerClient::check_message_size(1, Some(LINK_MAX_SIZE)) + .expect("a message below the link maximum keeps the current behavior"); + ProducerClient::check_message_size(LINK_MAX_SIZE - 1, Some(LINK_MAX_SIZE)) + .expect("a message below the link maximum keeps the current behavior"); + } + + // This differs on purpose from `create_batch`, which treats `None` as an + // error. AMQP 1.0 section 2.7.3 gives an unset or zero maximum the meaning + // "no limit", so the send path must not invent a limit of its own. + #[test] + fn message_size_is_not_checked_when_the_link_reports_no_maximum() { + ProducerClient::check_message_size(4 * LINK_MAX_SIZE, None) + .expect("with no link maximum the check is skipped, per AMQP 1.0 section 2.7.3"); + ProducerClient::check_message_size(u64::MAX, None) + .expect("with no link maximum the check is skipped, per AMQP 1.0 section 2.7.3"); + } } diff --git a/sdk/eventhubs/azure_messaging_eventhubs/tests/eventhubs_producer.rs b/sdk/eventhubs/azure_messaging_eventhubs/tests/eventhubs_producer.rs index d32596f51d2..7d43d62c300 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/tests/eventhubs_producer.rs +++ b/sdk/eventhubs/azure_messaging_eventhubs/tests/eventhubs_producer.rs @@ -686,3 +686,65 @@ async fn create_batch_rejects_size_above_link_maximum( Ok(()) } + +/// An event larger than the sender link allows must be refused, not sent. +/// +/// Both public entry points must refuse it, and the link must stay usable. +#[recorded::test(live)] +async fn send_event_rejects_message_above_link_maximum( + ctx: TestContext, +) -> Result<(), Box> { + use azure_messaging_eventhubs::models::{AmqpMessage, EventData}; + + // The size the live reproduction of issue #5101 used, against an Event Hubs + // link maximum of 1048576 bytes. + const TOO_LARGE_BODY: usize = 2 * 1024 * 1024; + + let recording = ctx.recording(); + let host = env::var("EVENTHUBS_HOST")?; + let eventhub = env::var("EVENTHUB_NAME")?; + + let client = ProducerClient::builder() + .with_application_id("send_event_rejects_message_above_link_maximum".to_string()) + .open(host.as_str(), eventhub.as_str(), recording.credential()) + .await?; + + let error = client + .send_event( + EventData::builder() + .with_body(vec![b'x'; TOO_LARGE_BODY]) + .build(), + None, + ) + .await + .err() + .expect("an event above the link maximum must be refused"); + assert!( + matches!(error.kind, ErrorKind::MessageSizeExceeded { .. }), + "send_event must report the message size kind, got: {error:?}" + ); + info!("send_event refused the large event: {error}"); + + let error = client + .send_message( + AmqpMessage::builder() + .with_body(vec![vec![b'x'; TOO_LARGE_BODY]]) + .build(), + None, + ) + .await + .err() + .expect("a message above the link maximum must be refused"); + assert!( + matches!(error.kind, ErrorKind::MessageSizeExceeded { .. }), + "send_message must report the message size kind, got: {error:?}" + ); + + // The refusal applies to the one large message. A normal event on the same + // client must still go, which also shows the link is still up. + client.send_event("Hello, Event Hub!", None).await?; + + client.close().await?; + + Ok(()) +} From 2d42a8ea75478dccb793c698857a22d96df9aa20 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 20 Aug 2026 14:49:54 -0400 Subject: [PATCH 2/2] fix(eventhubs): enforce the link maximum on single event sends `send_event` and `send_message` did not compare the encoded message against the maximum the sender link reports. fe2o3-amqp treats that maximum as a split boundary: it fragments an oversized payload across transfer frames with `more = true` instead of refusing it, so a 2 MiB event reached the partition. The behavior is the same in 0.17.0, so an upgrade does not help and this client must make the check itself. `send_message` now reads the link maximum, encodes the message, and refuses it when the encoded size is larger. The boundary is inclusive, so a message of exactly the maximum is still sent. A link that reports no maximum is not checked, because AMQP 1.0 gives an unset or zero `max-message-size` the meaning "no limit". The batch path already enforced the same limit and does not change. The new `ErrorKind::MessageSizeExceeded { requested, max_allowed }` variant lets a caller branch on the kind instead of the message. It mirrors the `EventHubsException` with `FailureReason.MessageSizeExceeded` that .NET reports for the same message. Fixes #5101 --- .../azure_messaging_eventhubs/CHANGELOG.md | 2 ++ .../azure_messaging_eventhubs/src/error.rs | 23 +++++++++++++ .../src/producer/mod.rs | 34 ++++++++++++++++++- 3 files changed, 58 insertions(+), 1 deletion(-) diff --git a/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md b/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md index 4696c059971..b8a3ebd3c41 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md +++ b/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md @@ -12,6 +12,7 @@ - The `EventProcessor` now opens every partition receiver with AMQP epoch (owner level) `0` and surfaces broker-initiated displacement as the new `EventHubsError::ConsumerDisconnected` error kind. When a second `EventProcessor` instance claims a partition this instance is currently holding, the broker disconnects this instance's receiver and the consumer's `stream_events()` resolves with `ConsumerDisconnected`. This matches the behavior of `EventProcessorClient` in the .NET and Java Azure SDKs. Consumers should pattern-match on `ErrorKind::ConsumerDisconnected` to detect a stolen partition and re-acquire a client via `next_partition_client()`. - Added `EventHubsError::ConsumerDisconnected(Option)` error variant. - Added the `ErrorKind::InvalidBatchSize { requested, max_allowed }` error variant. `create_batch` reports it when `EventDataBatchOptions::max_size_in_bytes` is zero or is larger than the maximum the sender link allows, so a caller can branch on the kind instead of the message. This matches the `ArgumentOutOfRangeException` that .NET raises and the typed error that Go returns for the same input. +- Added the `ErrorKind::MessageSizeExceeded { requested, max_allowed }` error variant. `send_event` and `send_message` report it when the encoded message is larger than the maximum the sender link allows, so a caller can branch on the kind instead of the message. This matches the `EventHubsException` with `FailureReason.MessageSizeExceeded` that .NET reports for the same message. ### Breaking Changes @@ -32,6 +33,7 @@ - Fixed a deadlock when a CBS failure during management-client creation started connection recovery. ([#4728](https://github.com/Azure/azure-sdk-for-rust/issues/4728)) - Closed a stale-resource window in connection recovery. A `ReconnectConnection` recovery that fired while a slow-path attach (authorize, session begin, or sender/receiver link attach) was in flight could cache a resource bound to the just-dropped connection; the next operation on that resource failed (unauthorized / detached / closed) and triggered a second, redundant recovery cycle. A recovery generation counter now tags each cached resource, and a slow path that completes across a recovery discards its result and re-attaches against the new connection instead of caching the stale one. The authorizer's token cache is mutable (a background task refreshes tokens) so it cannot use the same one-shot cell as the connection caches; both of its writers, `authorize_path` and the refresh task, instead re-check the generation under the same lock that recovery's clear takes, and a recovery brackets its invalidation with a generation bump on each side, which leaves the counter odd for as long as the recovery runs, so a slow path that overlaps a recovery at either end also discards rather than caching a resource bound to the connection that recovery is dropping. A token refresh pass that a recovery discards now applies the same backoff floor as a failed pass, so a recovery storm cannot turn the refresh loop into an uncapped stream of credential and CBS calls. The per-path / per-partition concurrency is preserved. ([#4454](https://github.com/Azure/azure-sdk-for-rust/issues/4454)) - `InMemoryCheckpointStore` now rotates the ETag and refreshes `last_modified_time` when an existing ownership is renewed, matching the create path and the production `BlobCheckpointStore`. Previously the renewal path reinserted the caller's record verbatim, leaving a stale ETag and timestamp; that divergence from the real store could mask bugs in code that relies on ETag rotation for optimistic concurrency. ([#4594](https://github.com/Azure/azure-sdk-for-rust/issues/4594)) +- `send_event` and `send_message` now reject a message larger than the maximum the sender link allows, and they do not transfer it. The AMQP library treats that maximum as a split boundary and fragmented an oversized message across transfer frames, so a 2 MiB event reached the partition. The batch path already enforced the same limit. ([#5101](https://github.com/Azure/azure-sdk-for-rust/issues/5101)) ### Other Changes diff --git a/sdk/eventhubs/azure_messaging_eventhubs/src/error.rs b/sdk/eventhubs/azure_messaging_eventhubs/src/error.rs index 1011b14f5cc..7309c42e987 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/src/error.rs +++ b/sdk/eventhubs/azure_messaging_eventhubs/src/error.rs @@ -40,6 +40,20 @@ pub enum ErrorKind { max_allowed: u64, }, + /// The encoded message is larger than the maximum the sender link + /// allows. The message was not sent. + /// + /// Mirrors the `EventHubsException` with + /// `FailureReason.MessageSizeExceeded` that .NET reports for the same + /// message. Match on the variant to tell it apart from a transport + /// failure: `matches!(err.kind, ErrorKind::MessageSizeExceeded { .. })`. + MessageSizeExceeded { + /// The encoded size of the message in bytes. + requested: u64, + /// The largest message size in bytes the sender link allows. + max_allowed: u64, + }, + /// Represents the source of the AMQP error. /// This is used to wrap an AMQP error in an Even Hubs error. /// @@ -103,6 +117,15 @@ impl std::fmt::Display for EventHubsError { It must be from 1 to {} bytes, which is the maximum the sender link allows.", requested, max_allowed ), + ErrorKind::MessageSizeExceeded { + requested, + max_allowed, + } => write!( + f, + "The message is {} bytes, which is larger than the {} bytes \ + the sender link currently allows.", + requested, max_allowed + ), ErrorKind::SendRejected(e) => write!(f, "Send rejected: {:?}", e), ErrorKind::InvalidManagementResponse => f.write_str("Invalid management response"), ErrorKind::AmqpError(source) => write!(f, "AMQP Error: {:?}", source), diff --git a/sdk/eventhubs/azure_messaging_eventhubs/src/producer/mod.rs b/sdk/eventhubs/azure_messaging_eventhubs/src/producer/mod.rs index d9539fbb5ae..88199502d05 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/src/producer/mod.rs +++ b/sdk/eventhubs/azure_messaging_eventhubs/src/producer/mod.rs @@ -8,7 +8,7 @@ use crate::{ recoverable::{RecoverableConnection, RecoverableSender}, ManagementInstance, }, - error::Result, + error::{ErrorKind, Result}, models::{AmqpMessage, EventData, EventHubPartitionProperties, EventHubProperties}, EventHubsError, RetryOptions, }; @@ -182,6 +182,9 @@ impl ProducerClient { /// Note: /// - If the event being sent does not have a message ID, a new message ID will be generated. /// - If the event options contain a partition ID, the event will be sent to the specified partition. + /// - If the encoded event is larger than the maximum the sender link allows, + /// the event is not sent and the error kind is + /// [`ErrorKind::MessageSizeExceeded`]. /// pub async fn send_event( &self, @@ -211,6 +214,10 @@ impl ProducerClient { /// /// Note: /// - The message is sent to the service unmodified. + /// - If the encoded message is larger than the maximum the sender link allows, + /// the message is not sent and the error kind is + /// [`ErrorKind::MessageSizeExceeded`]. + /// - A sender link that reports no maximum is not checked. /// #[tracing::instrument( level = "debug", @@ -241,6 +248,11 @@ impl ProducerClient { } let sender = self.connection.get_sender(target.clone()).await?; + let message: AmqpMessage = message.into(); + let link_max_size = sender.max_message_size().await?; + let encoded_size = AmqpMessage::serialize(&message)?.len() as u64; + Self::check_message_size(encoded_size, link_max_size)?; + let outcome = sender .send( message, @@ -292,6 +304,26 @@ impl ProducerClient { } } + /// Makes sure the encoded message fits the maximum the sender link reports. + /// + /// The boundary is inclusive: a message of exactly the maximum is sent. + /// fe2o3-amqp splits an oversized payload across transfer frames instead + /// of refusing it, so this client must make the check itself. A link that + /// reports no maximum is not checked: AMQP 1.0 gives an unset or zero + /// `max-message-size` the meaning "no limit", and fe2o3-amqp maps a zero + /// to `None`. + pub(crate) fn check_message_size(encoded_size: u64, link_max_size: Option) -> Result<()> { + match link_max_size { + Some(max_allowed) if encoded_size > max_allowed => { + Err(EventHubsError::from(ErrorKind::MessageSizeExceeded { + requested: encoded_size, + max_allowed, + })) + } + _ => Ok(()), + } + } + const BATCH_MESSAGE_FORMAT: u32 = 0x80013700; /// Creates a new batch of events to send to the Event Hub.