fix(eventhubs): discard stale resources across connection recovery

[Johnathan W (j7nw4r)]

Fixes #4454.

Summary

Four slow paths in RecoverableConnection and Authorizer share the same shape: a brief lock for a cache lookup, then AMQP IO (authorize / session begin / link attach) with no lock held, then a brief lock to insert. If a ReconnectConnection recovery fires during that lock-free IO window, it clears the caches, and the slow path then inserts a resource bound to the now-dead connection.

The four racing call sites:

• Authorizer::authorize_path
• RecoverableConnection::get_session, ensure_sender, ensure_receiver

Motivation

The next fast-path lookup hands out the stale resource; its first operation fails (unauthorized / detached / closed), the retry loop recovers a second time, and it eventually stabilizes. The path is self-healing, but every reconnect that races an in-flight slow path costs an extra recovery cycle, which shows up as brief unauthorized/detached errors right after a recovery and occasional flakes in forced-reconnect tests.

Changes

A recovery generation counter closes the window.

RecoverableConnection gains an AtomicU64 generation, read as a sequence lock. apply_recovery_plan bumps it once before it invalidates anything and once after the last invalidation. The value therefore changes across every recovery, and it is odd for exactly as long as one is in flight. generation_is_current(captured) holds only when captured is even and still equals the current value. Each cached cell is tagged with the generation it was created under (the per-path maps now hold a small GenerationalCell<T> wrapper instead of a bare Arc<OnceCell<Arc<T>>>).

Both halves of that test are necessary. Equality alone accepts an attach that starts and finishes inside a recovery, because the generation has not moved since the capture. That span is not short: apply_recovery_plan releases each lock before it takes the next, so a contended recovery can stall between its bumps while another task holds a cache lock. A slow path that captured a generation there has that whole time to attach to the connection the recovery is dropping. The parity half rejects it.

The three connection slow paths funnel through one helper, get_or_init_generational. The helper captures the generation, resolves the cell, runs the attach without the map lock, then tests the cell's generation. If the cell is current, the result is fresh and is returned and cached. If it is not, the just-attached resource is bound to a torn-down connection, so the helper evicts the stale cell and attaches again against the new generation. MAX_GENERATION_RETRIES bounds the loop, and every generation-guarded cache in the crate shares that one constant.

The authorizer's token cache is mutable (a background task refreshes entries), so it cannot use a OnceCell. Both of its writers apply the same test by hand, against the same authorization_scopes write lock that recovery's clear() takes:

• authorize_path captures the generation before the lock-free CBS attach and tests it inside the write lock, immediately before the insert. A recovery therefore cannot land between the test and the insert. On a rejection it drops the token and authorizes again against the new generation (bounded retries).
• The background refresh task is the cache's second writer. It captures the generation before its lock-free re-authorizations and tests it under the write lock before it writes the refreshed tokens back. On a rejection it discards them instead of repopulating the just-cleared cache. Its single-pass body is extracted as refresh_due_tokens so the guard is unit-testable.

A discarded refresh pass returns a distinct RefreshPass::Discarded, and the caller applies the same backoff floor it applies to a failed pass. Nothing failed in that case, but the cache keeps the old tokens. They stay due, the top-of-loop sleep is skipped, and the next pass starts at once. ReconnectSession and ReconnectLink advance the generation and leave the token cache populated, so without the floor a recovery storm turns the refresh loop into an uncapped stream of credential and CBS calls. That is the hazard PR #4593 installed the floor for.

The fast path in authorize_path returns a cached token without a generation test, which is deliberate. ReconnectSession and ReconnectLink advance the generation while the connection, and therefore its CBS authorization, survives. Tagging tokens with the generation would force a needless re-authorization of every path after every link recovery. Both production callers already sit inside a guard: ensure_sender and ensure_receiver call authorize_path inside the get_or_init_generational closure, so an attach that used a token from a torn-down connection is discarded and retried by the enclosing guard, and the management client builds inside recover_azure_operation.

A single counter is used for all resource types. Session-level recovery is rare, so the occasional extra re-init of an unaffected resource type after a narrower recovery is cheaper than the bookkeeping of per-type counters. This is noted inline in the generation field's comment.

The issue's single-Mutex<ConnectionState> sketch no longer applies: it predates #4446/#4517, which already moved the caches to per-path locks to cut contention, and collapsing back to one mutex would undo that. The per-path design is retained and only the generation-counter half of the sketch is applied.

Test plan

Deterministic unit tests in connection.rs:

• simulate_reconnect_bumps_generation asserts a simulated ReconnectConnection advances the generation by two, which pins the bracketing.
• recovery_generation_differs_for_a_mid_recovery_capture holds the token cache's write lock to stop a recovery inside authorizer.clear(), captures the generation there, and asserts it differs once the recovery completes.
• generation_captured_mid_recovery_is_never_current parks the counter mid-recovery and asserts a capture taken there is never current. It then asserts an attach made entirely inside a recovery retries to the budget and reaches neither the caller nor the cache.
• stale_generation_cell_is_replaced asserts a cell resolved at one generation is discarded and replaced by a fresh, distinct cell after a recovery, and that it stays stable within a generation.
• get_or_init_generational_discards_value_produced_during_recovery drives the helper with an init closure that fires a reconnect mid-attempt. It asserts the stale value is thrown away, the closure runs again exactly once, and the cached cell holds the fresh value.
• or_init_cell_reuses_newer_cell_and_replaces_older and get_or_init_generational_returns_superseding_peer_cell cover supersession: a caller that captured an older generation must not clobber a peer's newer cell.

Deterministic unit tests in authorizer.rs:

• authorize_path_discards_token_authorized_during_recovery uses a gated credential to block the first get_token precisely inside the lock-free authorization window, advances the generation there, and releases the block. It asserts authorize_path discards the stale token, authorizes again exactly once, and caches the fresh token at the post-recovery generation with no second recovery cycle.
• refresh_discards_tokens_refreshed_during_recovery seeds the cache with a token that is already due, then drives the extracted refresh_due_tokens with a gated credential that blocks inside the lock-free refresh window and advances the generation there. It asserts the pass returns RefreshPass::Discarded, which is the signal that makes the caller back off, and that the refreshed token is discarded while the original stays intact.

Live test:

• producer/mod.rs: force_errors_concurrent_authorize_send_reconnect extends the existing force_errors harness. It derives the partition IDs from the Event Hub (via get_eventhub_properties) rather than hard-coding "0".."3", asserts at least two exist, and sends to up to four of them concurrently across a forced ConnectionClosedByRemote. Several authorize_path re-authorizations are therefore in flight when the connection is torn down. A token cached against the dead connection would surface as an unauthorized/detached error and fail a send loop. A clean 30s pass means every partition authorized again against the new connection without a second recovery cycle.

The deterministic tests use atomic-flag gating (not timing) to interleave the recovery, and they were re-run to make sure they are not flaky.

Local checks (all green): cargo test --lib (135 passed), cargo clippy --all-features --all-targets -- -D warnings, cargo fmt --check, cargo doc --no-deps --all-features, all -p azure_messaging_eventhubs.

[Copilot]

Pull request overview

This PR fixes a race in azure_messaging_eventhubs connection recovery where slow-path AMQP attaches (authorize / session begin / link attach) could complete after a reconnect and incorrectly cache resources bound to the old, torn-down connection—causing a redundant second recovery cycle and brief unauthorized/detached/closed errors.

Changes:

• Add a recovery generation counter to RecoverableConnection and tag per-path cached cells so stale in-flight attaches are discarded and re-initialized against the new generation.
• Apply the same generation-guard pattern to Authorizer::authorize_path and the token refresh task to prevent stale token entries from being cached across recovery.
• Add deterministic unit tests for generation bumping and stale discard behavior, plus a new live test covering concurrent re-authorization during forced reconnect; update changelog.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 6 comments.

File	Description
sdk/eventhubs/azure_messaging_eventhubs/src/producer/mod.rs	Adds a live stress test that concurrently sends to multiple partitions during forced reconnect to catch stale-token/resource regressions.
sdk/eventhubs/azure_messaging_eventhubs/src/common/recoverable/connection.rs	Introduces generation-tagged per-path caches and a guarded get_or_init_generational helper to discard stale resources created during recovery.
sdk/eventhubs/azure_messaging_eventhubs/src/common/authorizer.rs	Adds generation-guarded authorization and refresh write-back to avoid caching tokens authorized/refreshed on a torn-down CBS link.
sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md	Documents the stale-resource recovery fix in the unreleased changelog.

[Sagar Patel (sagar0207)]

Approving. Looks good to me.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (2)

sdk/eventhubs/azure_messaging_eventhubs/src/common/recoverable/connection.rs:1042

• The generation bump is not synchronized with authorization_scopes, so the authorizer still has a stale-token window. After this increment but before clear() acquires its write lock, authorize_path can take its fast path and return a pre-recovery cache entry while treating the new generation as current; conversely, recovery can increment after a slow writer's generation check while that writer holds the lock, allowing it to return a token authorized on the dropped CBS connection. This can still produce the unauthorized/detached operation and redundant recovery cycle this PR is intended to eliminate. Make the generation transition and authorizer clear one write-lock critical section (and order connection teardown around that transition), or generation-tag token entries so reads can reject old entries.

            self.generation.fetch_add(1, Ordering::AcqRel);
        }

        if plan.clear_authorizer {
            self.authorizer.clear().await;

sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md:20

• This PR changes only private recovery internals and tests, with no public API change. Coding guideline 1000002 says not to modify a CHANGELOG for such a change unless adding a new unreleased version or dating an Unreleased section; neither exception applies here, so remove this entry.

- 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 the generation is bumped before that clear so an in-flight authorization observes the recovery and discards rather than repopulating the just-cleared cache. The per-path / per-partition concurrency is preserved. ([#4454](https://github.com/Azure/azure-sdk-for-rust/issues/4454))

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (2)

sdk/eventhubs/azure_messaging_eventhubs/src/common/authorizer.rs:1113

• Use “a OnceCell” rather than “an OnceCell”.

    // an `OnceCell` like the connection caches; `authorize_path` guards itself with

sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md:20

• This PR changes only internal recovery behavior and does not add or modify a public API. Per the changelog guideline, CHANGELOG.md should not be changed for an internal-only fix in an existing unreleased section; please remove this entry.

- 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 the generation is bumped before that clear so an in-flight authorization observes the recovery and discards rather than repopulating the just-cleared cache. The per-path / per-partition concurrency is preserved. ([#4454](https://github.com/Azure/azure-sdk-for-rust/issues/4454))

[Heath Stewart (heaths)]

Don't know a lot about EH recovery, but the solution does make some sense: multiple generations may happen depending on timing, and each generation has the potential of unique connections, right? So grouping them with atomic monoatomic generations make sense.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (3)

sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md:20

• This PR changes internal recovery behavior but does not change a public API, so it should not add an entry to the existing unreleased changelog section. Please remove this entry.

- 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 the generation is bumped before that clear so an in-flight authorization observes the recovery and discards rather than repopulating the just-cleared cache. The per-path / per-partition concurrency is preserved. ([#4454](https://github.com/Azure/azure-sdk-for-rust/issues/4454))

sdk/eventhubs/azure_messaging_eventhubs/src/common/recoverable/connection.rs:1055

• The connection is removed before the generation is advanced. On a multi-thread runtime, an in-flight initializer can finish on another worker in that interval, observe the old generation, and return the resource bound to the connection that was just removed. Acquire the connection mutex first, then advance the generation and take the connection while that mutex is still held; this prevents new initializers from obtaining the old connection after the bump and makes existing initializers fail the generation check.

        if plan.drop_connection {
            self.connections.lock().await.take();
            debug!(connection_id = %connection_id, "Recovery: dropped AMQP connection.");
        }

sdk/eventhubs/azure_messaging_eventhubs/src/common/authorizer.rs:577

• A generation mismatch discards all refreshed tokens but still reports failed: false. For session/link recoveries the token cache is not cleared, so the due token keeps its old expiry and the outer loop immediately starts another credential/CBS refresh with no backoff. Repeated recoveries can therefore create an unbounded refresh storm; mark this pass as failed so the existing retry backoff is applied.

                if connection.generation() != captured {
                    debug!(
                        "Discarding tokens refreshed during recovery (#4454); the recovery generation advanced mid-refresh."
                    );

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (2)

sdk/eventhubs/azure_messaging_eventhubs/src/producer/mod.rs:1026

• This setup does not deterministically put authorize_path in flight when the error is forced. The 10-second warmup normally populates every sender and authorization cache; the injected error is then consumed by one send, and recovery bumps/clears the caches before post-recovery re-authorization begins. Consequently this can pass even if an authorization started before recovery is cached afterward. Gate an authorization slow path and inject the reconnect only after that gate is entered, as the unit test does, so the live test actually exercises the stated race.

            Duration::seconds(10), // Seconds until forcing the error.
            Duration::seconds(30), // Seconds until test timeout.

sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md:20

• This changelog entry is an implementation narrative rather than a concise release note. Condense it to the user-visible bug fix; the generation/locking details belong in the PR description.

- 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 the generation is bumped before that clear so an in-flight authorization observes the recovery and discards rather than repopulating the just-cleared cache. The per-path / per-partition concurrency is preserved. ([#4454](https://github.com/Azure/azure-sdk-for-rust/issues/4454))

[Josue Gomez (SwayGom)]

Reviewed by reading the head-SHA versions of connection.rs and authorizer.rs in full, plus claims_based_security.rs, management.rs, recoverable/receiver.rs, and the azure_core_amqp session/sender/receiver types. Did not build or run the suite — everything below is from source.

The core mechanism is correct. I went looking specifically for the ways this kind of change usually breaks, and it holds up:

• Bump-before-clear is implemented as claimed. The fetch_add at connection.rs:1054 precedes all four clears (1057 / 1061 / 1067 / 1073), and the guard at 1049-1053 covers every flag, so no plan can clear without bumping.
• Atomic orderings are sound — no Relaxed anywhere. The only load is Acquire (connection.rs:595); every store is AcqRel. For the authorizer's in-lock re-check the happens-before edge actually comes from the authorization_scopes RwLock, so Acquire is strictly safe; for the lock-free re-check at :648 there's no edge at all and correctness rests on single-location coherence, which is fine and which SeqCst wouldn't improve.
• The re-check really is in the same critical section as the insert. authorizer.rs:190-201 and refresh_due_tokens at :572-584 both acquire the write guard, compare synchronously, and insert with the guard still live. No .await in either gap.
• Bounded retries never return a stale resource. MAX_GENERATION_RETRIES = 8 in both places; both loops fall through to Err.
• GenerationalCell has no TOCTOU. generation is immutable per cell and read from an owned clone taken under the map lock, so the tag can't tear against the contents. Comparing entry.generation rather than the captured local at :648 is the right choice — comparing the captured value would evict a peer's valid work.
• No new deadlock or lock-order inversion. get_or_init_generational holds no lock across init; the eviction block has no .await inside the write guard; apply_recovery_plan takes maps in a fixed order. The ensure_sender → get_session nesting uses the same key but different maps, so it isn't a re-entrant OnceCell init.
• The tests are genuinely deterministic and each fails without the fix. All seven gate on atomic flags with yield_now(), no sleeps. I traced each one for "would this pass unguarded?" and they wouldn't. I also checked authorize_path_discards_token_authorized_during_recovery's calls == 2 for a third-call race from the spawned refresher — the 1-hour expiry vs. the 6-minute TOKEN_REFRESH_BIAS means it sleeps ~54 min on the first pass, so it's stable.

That's a well-built change. The issues below are around the edges of it.

---

🔴 Issue 1 (Major) — a discarded refresh pass reports failed: false, so the refresh loop re-runs with no backoff

authorizer.rs:574-589, consumed at :404-420.

When the generation guard discards a refresh, refresh_failed is never set, so refresh_due_tokens returns Completed { failed: false } and refresh_tokens takes the empty {} arm — no backoff. But the discarded tokens leave every due path's expires_on unchanged, so the next iteration's refresh_time is still in the past, the top-of-loop sleep is skipped (:398-400), and the pass immediately re-runs a full get_token + perform_authorization for every due path.

This is the exact hazard already identified and fixed for the sibling path. From authorizer.rs:28-31:

TOKEN_REFRESH_RETRY_BACKOFF — "A failed refresh leaves the token's expires_on unchanged, so its computed refresh_time stays in the past and the next pass would not sleep. Without this floor a persistent failure ... busy-spins, hammering get_token / perform_authorization with no backoff. See PR #4593 review."

The new discard branch has an identical property and no floor.

Reachability is good because of the RecoveryPlan table at connection.rs:230-247: ReconnectSession and ReconnectLink both set clear_authorizer: false while still bumping the generation. So link-level recoveries — the common ones — bump the counter but leave the token cache populated and still due, meaning the loop can't self-limit by emptying the cache the way ReconnectConnection does. For as long as a recovery storm (cluster upgrade, failover, flapping connection) keeps bumping, the refresh task issues an uncapped, un-delayed stream of AAD get_token calls and CBS round trips.

And it does so silently — the discard logs at debug!, and because refresh_failed stays false the warn! never fires.

Suggested fix: set refresh_failed = true on the discard path, or add a distinct RefreshPass::Discarded variant the caller sleeps on.

🔴 Issue 2 (Major, design call) — budget exhaustion turns a transient storm into a non-retryable user-visible failure

connection.rs:669-673, authorizer.rs:222-226.

The counter is a single global AtomicU64 per connection and every plan bumps it, including per-link ReconnectLink. On a many-partition ProducerClient/EventProcessor, an unrelated partition's link error invalidates every other partition's in-flight attach. Eight such bumps inside one attach window is plausible during an upgrade or failover, and there's no backoff between the 8 retries — each iteration goes straight back to a full attach.

On exhaustion the code returns AmqpError::with_message(...), which should_retry_amqp_error's _ arm (connection.rs:1296-1299) classifies as ReturnError — non-retryable. That surfaces to the caller: send_batch hard-fails, stream_events() terminates that partition processor.

Before this PR the same scenario was self-healing (cache the stale resource → next op fails with a retryable transport error → recover). The in-code comment justifies ReturnError as preventing "spinning forever," but that conflates two things: the loop already can't spin without forward progress, since each failed iteration requires a real generation bump and entry.generation <= current_generation() always holds. The runaway risk is a red herring; the cost is a transient converted to fatal.

Worth an explicit decision from the owners rather than a silent one. Options: scope the counter per resource type or per path, or add backoff between iterations and classify the exhaustion error as retryable so the outer recover_azure_operation handles it.

🟡 Issue 3 (Minor) — drop_connection runs before the bump, so the stated invariant isn't fully held

connection.rs:1031-1055. The connection is taken and dropped at 1031-1034, but the bump is at 1054. In that window the connection is gone while the caches still validate as current-generation. Two effects: a fast path can hand out a resource whose connection is already dropped; and a slow path that captured the pre-bump generation will call ensure_connection(), re-create the connection, attach a valid resource against it, then be forced to discard it when the bump lands — wasting an attach and consuming retry budget (feeding Issue 2).

Both windows are strictly narrower than pre-PR behaviour, so this is residual rather than a regression — but the description's framing implies a stronger guarantee than the code provides. Moving the fetch_add above the drop_connection block is safe, since drop_connection: true always implies the clear flags.

🟡 Issue 4 (Minor) — or_init_cell stamps a fresh cell with the captured generation, guaranteeing a wasted attach

connection.rs:184-192. When creating a fresh cell it uses the generation argument captured at :629 before the lock, rather than re-reading under the write lock it already holds. If a recovery lands in that window the new cell is stamped stale and the check at :648 is guaranteed to fail — even though the resource was attached against the live post-recovery connection. Deterministic wasted attach plus one consumed retry slot. Re-reading self.current_generation() inside the write-lock section is strictly better and can't over-stamp, since the cell is empty at that point.

🟡 Issue 5 (Minor) — the refresh pass now holds a strong Arc across the whole pass

authorizer.rs:500-586. The Weak::upgrade moved out of the per-path loop (strong Arc dropped each iteration) up to the top of the pass, where it's held across every path's get_token + perform_authorization and the write-back. ProducerClient::close() (producer/mod.rs:158) uses Arc::try_unwrap and fails with "Could not close ... multiple references exist" if any strong ref is outstanding — so the spurious-close window widens from one path's CBS attach to a full pass. Low probability given ~50-minute pass spacing, but a real behavioural regression from an otherwise-good refactor.

🟡 Issue 6 (Minor) — the two invariants called load-bearing are the two nothing pins

simulate_reconnect_bumps_generation only asserts the counter advanced. Moving the fetch_add at :1054 below the clear_authorizer block would reintroduce exactly the race this PR describes, and every test would still pass. Given the description calls that ordering load-bearing, it's the one mutation worth a test.

Separately, the retry-budget exhaustion path is untested in both loops. connection.rs:1832-1845 pins the classification of a with_message error, but nothing asserts either loop terminates at 8 and returns it.

---

Clean

No panic paths added to production code (the expects at :316, :345, :1099 are all #[cfg(test)]). No unsafe added — the unsafe impl Send/Sync at :150-151 is pre-existing diff context. No public API change; generation() is pub(crate), GenerationalCell/RefreshPass are private, and the three *_cell helpers were correctly narrowed to #[cfg(test)].

One note I chased and cleared: discarded resources are dropped without detach()/end() (:648-666), and there's no Drop impl in azure_core_amqp that detaches — Fe2o3AmqpSession::drop only logs. But this is not new: the existing clear() for ReconnectSession/ReconnectLink already drops live links without detaching while the connection stays up, and the discard-then-retry is serialized on one task, so an epoch receiver's retry attaches at the same epoch and steals its own orphan per the amqp:link:stolen semantics at recoverable/receiver.rs:181. Not an issue.

---

Merge coordination

This, #4806 and #4895 all branch from 97e0d63c and all modify the RecoverableConnection field list, the constructor, apply_recovery_plan, and mod tests in connection.rs. They will conflict regardless of order.

Two interactions worth naming:

1. This PR does not touch mgmt_client (still an AsyncMutex on this base), so after #4806 lands its new mgmt OnceCell would be the only cache without the generation check — i.e. the one remaining #4454 window. Someone needs to own wiring it into GenerationalCell.
2. Conversely, this PR's retry loop can run perform_authorization up to 8 times inside the mgmt_client mutex region that #4806 fixes, mildly amplifying that deadlock while it exists. Near-neutral in practice — the deadlock fires on the first re-entry — but it argues these two want to land close together.

I'd suggest this one first, then rebase #4806 onto it and make the mgmt cell generational in the same pass.

---

Bottom line: the mechanism is right and the tests are good. I'd want Issue 1 fixed before merge — it's a silent, un-backed-off credential-request loop under exactly the recovery-storm conditions this PR targets, and it contradicts a backoff floor this same file already established for the identical failure shape. Issue 2 deserves an explicit decision. The rest are polish.

[Josue Gomez (SwayGom)]

Adding the line-anchored version of Issue 1 from my earlier review, so it sits on the code rather than in a wall of text.

To be clear on severity: this is the one item I'd want fixed before merge. The rest of that review was polish, and the core generation mechanism verified correct.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (5)

sdk/eventhubs/azure_messaging_eventhubs/src/common/recoverable/connection.rs:663

• The equality check can succeed while recovery is still in progress. A task can capture the generation after the opening bump, attach against the old connection before it is dropped, and reach this check before the closing bump; it then returns the stale resource that this PR is intended to discard. Treat the bracketed state as explicitly in-progress (and ensure overlapping recoveries cannot make it appear stable), and only initialize/return resources from a stable generation.

            if self.current_generation() == entry.generation {
                return Ok(value.clone());

sdk/eventhubs/azure_messaging_eventhubs/src/common/authorizer.rs:162

• The cached-token fast path is unguarded. Between the opening recovery bump and clear(), a caller can read and return a token whose CBS authorization belongs to the connection being dropped; for example, after drop_connection but before clear_authorizer, this skips authorization on the newly opened connection. Validate that the generation is stable around this lookup or tag cached tokens with their generation instead of returning solely on cache presence.

            // Fast path: cached token under a brief lock.
            if let Some(token) = self.authorization_scopes.read().await.get(path).cloned() {
                debug!("Token already exists for path: {path}");
                return Ok(token);

sdk/eventhubs/azure_messaging_eventhubs/src/common/authorizer.rs:208

• This check can accept the opening generation while recovery is still between its two bumps. If authorization finishes on the old CBS link, recovery clears the cache, and this task resumes before the closing bump, it observes the same generation and reinserts the stale token after clear(); because token entries are not generation-tagged, the closing bump does not invalidate that entry. Reject an in-progress generation (with recovery serialization/counting for concurrent recoveries) before inserting.

            let stored = {
                let mut scopes = self.authorization_scopes.write().await;
                if connection.generation() != generation {
                    None
                } else {
                    // If another task won the race, return its cached token and drop
                    // ours. Both CBS auths succeeded against the same link, so either
                    // credential is acceptable to the broker.
                    Some(scopes.entry(path.clone()).or_insert(token).clone())

sdk/eventhubs/azure_messaging_eventhubs/src/common/authorizer.rs:609

• The refresh writer has the same mid-recovery hole: it can capture the opening generation, authorize on the old connection, resume after clear(), and write the stale tokens back before the closing bump. The equality test still passes, and the untagged token cache keeps those entries afterward. Only permit write-back from a stable generation, including correct handling of overlapping recoveries.

            if !updated_tokens.is_empty() {
                let mut scopes = self.authorization_scopes.write().await;
                if connection.generation() != captured {
                    debug!(
                        "Discarding tokens refreshed during recovery (#4454); the recovery generation advanced mid-refresh."
                    );
                    // Report the discard to the caller so it applies the backoff
                    // floor. The cache keeps the old tokens, so they are still due
                    // and the next pass would start with no delay (#4454).
                    return Ok(RefreshPass::Discarded);

sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md:21

• This PR changes only crate-internal recovery behavior and tests; it does not change a public API. Coding guideline 1000002 says not to modify a changelog when there is no public API change unless adding a new unreleased version or dating an existing Unreleased section, neither of which occurs here. Remove this entry.

- 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, so a slow path that starts part way through a recovery 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))

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (2)

sdk/eventhubs/azure_messaging_eventhubs/src/common/recoverable/connection.rs:1118

• The closing bump happens only after authorizer.clear() has released the token-cache write lock, leaving the stale-write window open. A new-path authorization can capture the opening generation, begin CBS authorization on the old connection, then finish after clear() but before this bump; its guarded insert sees the same generation and repopulates the cleared cache with the stale token. The refresh writer has the same interleaving, and this final bump does not evict or tag the inserted token, so later fast-path lookups keep returning it. Please represent the between-bumps state as recovery-in-progress and reject/wait on that state, or otherwise make each authorizer clear and the closing publication atomic with respect to both writers.

        // Closing bump. See the comment above the opening one.
        if invalidates {
            self.generation.fetch_add(1, Ordering::AcqRel);

sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md:21

• This PR changes only internal recovery behavior and does not change a public API. The changelog policy requires omitting changelog edits for such changes unless adding a new unreleased version or dating an Unreleased section, neither of which happens here; please remove this entry.

- 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, so a slow path that starts part way through a recovery 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))

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated no new comments.

Suppressed comments (3)

sdk/eventhubs/azure_messaging_eventhubs/src/common/recoverable/connection.rs:1128

• The odd/even sequence-lock invariant is not safe for this async recovery path. recover_from_error can run concurrently from independent sender/receiver retry loops, so two opening fetch_adds make the generation even while both recoveries are still invalidating state; a slow path can then pass generation_is_current and return a resource that one recovery subsequently tears down. Cancellation after this opening bump is also permanent: dropping an operation while an awaited clear is pending leaves the counter odd, causing every future generational initialization to exhaust its retry budget. Please serialize recoveries and make the closing transition cancellation-safe (for example, an async recovery mutex plus an RAII guard that restores the settled generation on drop).

        if invalidates {
            self.generation.fetch_add(1, Ordering::AcqRel);
        }

sdk/eventhubs/azure_messaging_eventhubs/src/common/recoverable/connection.rs:104

• The generation guard still omits the single mgmt_client cache, even though ensure_amqp_management has the same lock-free OnceCell initialization shape. If a full reconnect swaps the cell after an in-flight management attach succeeds but before that task returns, the task still returns the client from the old cell, bound to the dropped connection; its first call then triggers the redundant recovery this change is intended to eliminate. Please tag/check management-client initialization as well, or otherwise retry when its captured generation is superseded.

    sender_instances: RwLock<HashMap<Url, GenerationalCell<AmqpSender>>>,
    session_instances: RwLock<HashMap<Url, GenerationalCell<AmqpSession>>>,
    receiver_instances: RwLock<HashMap<Url, GenerationalCell<AmqpReceiver>>>,

sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md:22

• This changelog item is far too detailed for a release note. The changelog guideline requires concise, release-relevant entries; implementation details such as sequence-lock parity, cache writer locking, and refresh-loop backoff should remain in the PR description. Reduce this to a single-line summary of the user-visible recovery fix.

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