fix(eventhubs): box the receive timeout cause once

sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md

@@ -25,6 +25,7 @@
 - The `EventProcessor`'s load-balancer reconciliation now closes the underlying AMQP receiver for any partition that has been reassigned to another consumer, so the consumer's `stream_events()` resolves and the loop can terminate. Previously a stolen partition's client could continue to attempt receives until the broker tore down the link.
 - 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))
+- The error that a receive timeout produces now carries its cause unboxed, so `downcast_ref::<std::io::Error>()` returns the `std::io::Error` with `ErrorKind::TimedOut`. The cause was boxed twice, which stored a `Box<std::io::Error>` and made every downcast to `std::io::Error` return `None`. ([#5098](https://github.com/Azure/azure-sdk-for-rust/issues/5098))
 - `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))
 
 ### Other Changes

sdk/eventhubs/azure_messaging_eventhubs/src/common/recoverable/receiver.rs

@@ -41,6 +41,16 @@ impl RecoverableReceiver {
         RecoverableConnection::should_retry_receive_error(e)
     }
 
+    /// Builds the error that a receive timeout produces. The cause goes in
+    /// unboxed, because `azure_core::Error::new` boxes its argument and a
+    /// pre-boxed cause defeats `downcast_ref::<std::io::Error>()`.
+    fn receive_timeout_error() -> AmqpError {
+        AmqpError::from(azure_core::Error::new(
+            AzureErrorKind::Io,
+            std::io::Error::from(std::io::ErrorKind::TimedOut),
+        ))
+    }
+
     /// Wraps an `ensure_receiver` failure so the original error stays reachable
     /// through the source chain.
     ///
@@ -124,9 +134,7 @@ impl AmqpReceiverApis for RecoverableReceiver {
                     select! {
                         delivery = receiver.receive_delivery().fuse() => Ok(delivery),
                         _ = azure_core::sleep::sleep(delivery_timeout).fuse() => {
-                             Err(AmqpError::from(azure_core::Error::new(
-                                AzureErrorKind::Io,
-                                Box::new(std::io::Error::from(std::io::ErrorKind::TimedOut)))))
+                             Err(Self::receive_timeout_error())
                         },
                     }?
                 } else {
@@ -178,6 +186,19 @@ mod tests {
         )))
     }
 
+    // A caller that branches on a receive timeout downcasts the cause to
+    // `std::io::Error`. A cause that goes in already boxed is stored as
+    // `Box<std::io::Error>`, which no downcast to `std::io::Error` finds.
+    #[test]
+    fn receive_timeout_cause_downcasts_to_io_error() {
+        let error = azure_core::Error::from(RecoverableReceiver::receive_timeout_error());
+        assert_eq!(error.kind(), &AzureErrorKind::Io);
+        let cause = error
+            .downcast_ref::<std::io::Error>()
+            .expect("the receive timeout cause must downcast to std::io::Error");
+        assert_eq!(cause.kind(), std::io::ErrorKind::TimedOut);
+    }
+
     // The broker rejects a re-attach at the old epoch with `amqp:link:stolen`.
     // The wrapper must keep that condition reachable. Before the fix the
     // wrapper was `AmqpError::with_message`, which has no source, so the