diff --git a/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md b/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md index 8575377c254..70ca5f0ba41 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md +++ b/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md @@ -8,6 +8,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 @@ -26,6 +27,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 38c7f126592..95cd2b250c4 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. /// @@ -92,6 +106,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 7e8e14c8b9f..e8b33f211a9 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/src/producer/mod.rs +++ b/sdk/eventhubs/azure_messaging_eventhubs/src/producer/mod.rs @@ -6,7 +6,7 @@ use crate::{ recoverable::{RecoverableConnection, RecoverableSender}, ManagementInstance, }, - error::Result, + error::{ErrorKind, Result}, models::{AmqpMessage, EventData, EventHubPartitionProperties, EventHubProperties}, EventHubsError, RetryOptions, }; @@ -173,6 +173,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, @@ -202,6 +205,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", @@ -232,6 +239,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, @@ -283,6 +295,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. @@ -763,7 +795,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; use azure_core_test::{recorded, TestContext}; @@ -1188,4 +1222,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(()) +}