diff --git a/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md b/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md index 8575377c254..5de14ba89af 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md +++ b/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::()` returns the `std::io::Error` with `ErrorKind::TimedOut`. The cause was boxed twice, which stored a `Box` 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 diff --git a/sdk/eventhubs/azure_messaging_eventhubs/src/common/recoverable/receiver.rs b/sdk/eventhubs/azure_messaging_eventhubs/src/common/recoverable/receiver.rs index a4c19252802..657028c302a 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/src/common/recoverable/receiver.rs +++ b/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::()`. + 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`, 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::() + .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