fix(eventhubs): enforce the AMQP link maximum on single event sends

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<AmqpDescribedError>)` 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
 

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),

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<u64>) -> 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");
+    }
 }

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<dyn Error>> {
+    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(())
+}