fix(eventhubs): serialize claims-based-security authorizations

[Johnathan W (j7nw4r)]

Summary

Two claims-based-security authorizations that overlap on one AMQP connection make the link attach fail. The service permits only one $cbs link for each connection, and it rejects the second attach with NotAllowed ("A link to connection ... $cbs node has already been opened"). This change puts the authorizations of one connection in sequence.

Motivation

RecoverableConnection::ensure_amqp_cbs attaches a new $cbs link for each authorization, uses it, and then drops it. Nothing kept two of them apart. One authorization at a time is safe, because the previous link is gone before the next one attaches. Two authorizations that start at the same time are not.

The client classifies NotAllowed as not retryable, so the failure reaches the caller. The sender or receiver never attaches, and the operation fails.

A client that attaches one link at a time never sees this, which is why the fault stayed hidden. A client that sets up more than one link at once does see it. The failure rate depends on the timing, so a retry of the whole operation often succeeds and hides the cause.

Changes

• Added a cbs_lock to RecoverableConnection, and a lock_claims_based_security method that takes it.
• RecoverableClaimsBasedSecurity::authorize_path now holds that lock for the full round trip, which covers the link attach, the put-token, and the drop.
• Added the live test send_to_every_partition_at_once. It reads the partitions first, so the connection is open, and it then sends to every partition at the same time from one client. Only the sender attaches overlap.

The lock covers only the authorization. The link attach that follows, the session begin, and the sends and receives all stay concurrent, so this does not undo the per-path concurrency work of #4563.

Validation

cargo fmt --check, cargo clippy --all-targets --all-features -- -D warnings, cargo test --all-features (122 unit tests and 45 doc tests, no failures), and RUSTDOCFLAGS='-Dwarnings' cargo doc all pass.

The new test ran against a real Event Hub with 5 partitions, and it shows both states. With this change, 3 runs of 3 pass, and all 5 sends succeed in each run. With the lock removed and nothing else changed, 3 runs of 3 fail, at 3 of 5 sends each time. The trace of the failing runs gives the cause directly: NotAllowed, description: Some("A link to connection '268' $cbs node has already been opened.").

The live tests consumer_open_with_connection_string, send_eventdata_with_connection_string, and test_round_trip_connection_string also pass, so the single-authorization path does not regress.

The local test identity holds no Send role on the test namespace, so the new test ran through a local change that authenticates with a connection string. Its committed form uses recording.credential(), which matches the other tests in that file.

#4873 hits the same fault through a different path. It adds a client that starts one sender for each partition, and it carries this fix today. That branch rebases and drops the duplicate commit after this change merges.

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).
2 pipeline(s) were filtered out due to trigger conditions.
There may be pipelines that require an authorized user to comment /azp run to run.

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).
2 pipeline(s) were filtered out due to trigger conditions.
There may be pipelines that require an authorized user to comment /azp run to run.

[Copilot]

Pull request overview

Serializes CBS authorization round trips per AMQP connection to prevent overlapping $cbs link attachments.

Changes:

• Adds a connection-scoped CBS mutex.
• Holds the mutex through link attachment and authorization.
• Documents the bug fix.

Reviewed changes

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

File	Description
connection.rs	Adds the CBS authorization lock.
claims_based_security.rs	Serializes CBS round trips.
CHANGELOG.md	Adds the release note.

[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/CHANGELOG.md:17

• This release note repeats the implementation details and failure narrative rather than concisely summarizing the fix. Please shorten it in accordance with the changelog guideline.

- Claims-based-security authorizations for one connection now run in sequence. Each authorization attaches a `$cbs` link, uses it, and then drops it, but the service permits only one `$cbs` link for each connection. Two authorizations that started at the same time made the service reject the second one with `NotAllowed` ("A link to connection ... $cbs node has already been opened"), which the client classified as not retryable, so the link attach failed. A client that sets up more than one link at once hit this, for example a buffered producer that starts one sender for each partition.

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

• This says the authorizations run concurrently even though this method makes them sequential. Clarify that they can overlap only without the lock so the contract is unambiguous.

    /// full round trip. Authorizations for different paths run at the same time
    /// when a client sets up more than one link at once, for example a buffered
    /// producer that starts one sender for each partition.

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

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

[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 (1)

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

• This test does not detect moving or removing the guard in authorize_path; it only verifies that callers of lock_claims_based_security are mutually exclusive. Please describe the narrower guarantee so future maintainers do not mistake it for integration coverage of the guard's scope.

    // Regression guard for the `NotAllowed` failure: if a later change moves or
    // narrows the guard in `authorize_path`, two round trips can overlap again.
    // Count the callers that hold the lock at the same time, and make sure the
    // count never goes above one.

[Josue Gomez (SwayGom)]

The diagnosis and the fix are right — one authorization at a time is the correct semantics given the service permits a single $cbs link per connection, and the live-test evidence (3/3 pass with the lock, 3/3 fail at 3-of-5 sends without it, with the NotAllowed trace) is about as direct as it gets. Holding off on approval for one question.

Deadlock: checked, and it's clean. I specifically looked at whether this reintroduces the class of bug #4806 fixes, since it adds a lock held across an await in the same path that PR identifies as re-entrant. It does not — _cbs_guard is acquired inside the per-attempt async move closure, so it drops when that attempt's future completes, before recover_with_backoff awaits the recovery hook. And apply_recovery_plan never takes cbs_lock. Good.

---

The question: the lock is held across an unbounded await.

_cbs_guard covers ensure_amqp_cbs() plus the full authorize_path round trip, and recover_with_backoff imposes no per-attempt timeout:

loop {
    match operation().await {          // <- nothing bounds this
        Ok(result) => ...
        Err(err) => {
            let time_since_start = start_time.elapsed();
            if current_retry >= options.max_retries
                || time_since_start >= options.max_total_elapsed { ... }

max_total_elapsed is only evaluated after an attempt returns Err, so it never bounds a single attempt. If ensure_amqp_cbs() or the put-token stalls — a half-open TCP connection, a peer that accepts and stops responding — the guard is held indefinitely and every other authorization on that connection blocks behind it, forever.

That's a behavior change worth being explicit about: today a stalled CBS round trip blocks one authorization; after this change it stalls all of them on that connection. Note that #4806's own test harness constructs exactly this peer shape, so it isn't hypothetical.

I don't think this should block the fix — the status quo is a hard non-retryable NotAllowed, which is worse than a rare stall. But it converts a bounded failure into an unbounded one, so it deserves a decision rather than silence. Options, roughly in order of cost:

1. Wrap the guarded region in a timeout (the CBS round trip has a natural bound; on expiry drop the guard and let the retry loop handle it).
2. Fix recover_with_backoff to bound each attempt — broader blast radius, helps every caller, probably its own PR.
3. Document the tradeoff in the lock_claims_based_security doc comment so the next reader doesn't have to rediscover it.

Happy to approve on (1) or (3), or on a reasoned "the underlying ops can't stall indefinitely" if that's actually true — I couldn't establish it from this layer.

---

Minor: the unit tests don't test the fix. cbs_lock_blocks_a_second_caller_until_the_guard_drops and cbs_lock_never_lets_two_callers_overlap assert that async_lock::Mutex provides mutual exclusion — that's testing the dependency. The real coverage is send_to_every_partition_at_once, which is live and won't run in CI. Not asking for more here, since the failure mode is inherently service-side, but worth saying plainly in the description so the unit tests aren't read as covering the regression.

Merge coordination: this adds a field to RecoverableConnection and to its constructor, and appends to mod tests in connection.rs — same as #4806 and #4571, which both do the same in the same places off the same base (97e0d63c). All three will conflict. This one is the most orthogonal of the three, so it's probably cheapest to land last.

[Josue Gomez (SwayGom)]

Approving.

The diagnosis is correct, the fix matches the service's actual constraint (one $cbs link per connection), and the live-test evidence is about as direct as this class of bug allows — 3/3 pass with the lock, 3/3 fail at 3-of-5 sends without it, with the NotAllowed trace naming the cause. That's a definite, reproducible defect being closed.

I also verified the thing I was most worried about: this does not reintroduce the deadlock class #4806 fixes. _cbs_guard is acquired inside the per-attempt async move closure, so it drops when that attempt's future completes — before recover_with_backoff awaits the recovery hook — and apply_recovery_plan never takes cbs_lock. Clean.

On my earlier comment about the unbounded await: I'm treating that as a follow-up rather than a merge blocker. The reasoning is that the trade is clearly favourable — today's behaviour is a reproducible non-retryable NotAllowed that fails the user's operation, and the risk introduced is a rare, conditional stall that requires the CBS round trip to hang outright. Fixing a definite bug and leaving a conditional one is the right direction.

It does stay true that recover_with_backoff has no per-attempt bound (max_total_elapsed is only evaluated after an attempt returns Err), so a stalled CBS exchange now blocks every authorization on that connection rather than just its own. My ask is only that this not be invisible to the next reader — a sentence in the lock_claims_based_security doc comment noting the guarded region is unbounded would be enough, and can ride along in any later commit. Bounding attempts in recover_with_backoff is the better long-term fix but is genuinely a separate PR, since it changes behaviour for every caller.

One thing worth stating plainly in the description rather than leaving implied: the two new unit tests assert that async_lock::Mutex provides mutual exclusion, which tests the dependency rather than the fix. The real coverage is send_to_every_partition_at_once, and it's live, so it won't run in CI. That's a reasonable place to land given the failure is inherently service-side — just better said out loud than inferred.

Merge coordination (repeating from my earlier comment since it affects sequencing): this adds a field to RecoverableConnection and its constructor and appends to mod tests in connection.rs — the same three places #4806 and #4571 touch, all off base 97e0d63c. All three will conflict. This one is the most orthogonal, so it's cheapest to land last.

[Copilot]

Pull request overview

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

Suppressed comments (1)

sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md:17

• This PR changes internal synchronization but introduces no public API change. Guideline 1000002 says not to modify a changelog for such changes unless adding a new unreleased version or dating an existing Unreleased section; neither applies here, so remove this entry.

- Claims-based-security authorizations for one connection now run in sequence. The service permits one `$cbs` link for each connection, so a client that attached more than one link at once could fail with `NotAllowed`.

[Josue Gomez (SwayGom)]

The diagnosis here is excellent — the explanation of why this stayed hidden (single-link clients never overlap) and why retries mask it is very clear, and the NotAllowed trace is conclusive. I also checked the lock ordering, since that's the main hazard when introducing a mutex on this path: ensure_amqp_cbs → ensure_connection acquires connections while holding cbs_lock, and both authorize_path call sites (connection.rs:718, :778) let the ensure_connection guard drop before calling authorize_path, so nothing takes them in the reverse order. No reentrancy, no deadlock. Keeping the guard inside the recover_azure_operation closure is also the right call — the lock isn't held across retry backoff or connection recovery.

One issue I think should be addressed before merge.

cbs_lock_never_lets_two_callers_overlap doesn't test what its comment claims

// Regression guard for the `NotAllowed` failure: if a later change moves or
// narrows the guard in `authorize_path`, two round trips can overlap again.
#[tokio::test]
async fn cbs_lock_never_lets_two_callers_overlap() {
    ...
    let _guard = connection.lock_claims_based_security().await;

The test never calls authorize_path. It spawns 8 tasks that call lock_claims_based_security() directly and asserts they don't overlap.

That means if someone deletes the let _cbs_guard = ... line from authorize_path, this test still passes — it cannot detect the regression it's named for. What it actually asserts is that async_lock::Mutex provides mutual exclusion, which is the dependency's contract rather than something this crate needs to cover. The same applies to cbs_lock_blocks_a_second_caller_until_the_guard_drops, which exercises try_lock semantics.

I'd rather flag this than let it through, because the comment gives a future reviewer positive confidence that the invariant is guarded when it isn't — which is a worse position than having no test at all.

To actually guard the invariant, the test needs to drive authorize_path and assert non-overlap at the CBS boundary. This file already has a precedent for that kind of seam — #[cfg(test)] self.get_forced_attach_error()? inside ensure_receiver — so an analogous counting/instrumentation seam in ensure_amqp_cbs (increment on entry, decrement on exit, assert the max never exceeds 1) would let the test exercise the real call path without a network. If that turns out to be impractical, it would be worth softening the comment so it describes what's actually covered rather than implying authorize_path is protected.

[Josue Gomez (SwayGom)]

Re-approving to make the state unambiguous after my comment earlier today, and to consolidate it with my 7/30 review.

Position unchanged. The diagnosis is correct, the fix matches the service's actual constraint (one $cbs link per connection), and the live-test evidence is about as direct as this bug class allows — 3/3 pass with the lock, 3/3 fail at 3-of-5 sends without, with the NotAllowed trace naming the cause. I re-verified the deadlock question from a second angle since it's the main hazard when adding a mutex here: the lock ordering is cbs_lock → connections, and both authorize_path call sites (connection.rs:718, :778) release the ensure_connection guard before calling, so nothing acquires them in reverse. Combined with the guard being scoped inside the per-attempt closure, this does not reintroduce the class #4806 fixes.

Consolidating my comment from earlier today. It restated the point I'd already made on 7/30 — that the two unit tests assert async_lock::Mutex provides mutual exclusion rather than covering the fix. That was my oversight, apologies for the duplicate. I'm withdrawing it as an ask. The concrete suggestion in it stands only as an optional idea if someone wants it later: the #[cfg(test)] self.get_forced_attach_error()? seam in ensure_receiver is a precedent for an equivalent counting seam in ensure_amqp_cbs, which would let a test drive authorize_path and assert non-overlap without a network. Not a merge condition — the failure is inherently service-side and the live test is the honest coverage.

Two things I'd still like before this merges, both cheap:

1. Run the committed form of send_to_every_partition_at_once once. Per the description, validation used a local connection-string modification because the test identity holds no Send role, so the committed recording.credential() path has not actually been executed. That's unverified code in the PR, and it's a small fix for someone with the role.
2. The doc-comment sentence from my 7/30 review — noting that the guarded region is unbounded, so a stalled CBS exchange blocks every authorization on that connection rather than just its own. recover_with_backoff evaluates max_total_elapsed only after an attempt returns Err, so nothing bounds a single attempt. One sentence keeps the next reader from rediscovering it.

Explicitly not blocking on: bounding attempts in recover_with_backoff (separate PR — it changes behavior for every caller), caching the CBS client in a OnceCell so the collision becomes impossible by construction, or added unit coverage.

One thing worth a number rather than a change: a client opening N partitions now performs N sequential CBS round trips, which is the exact scenario this PR targets. Correctness was measured; open-time wasn't. If someone has a 32-partition hub handy, a before/after on open latency would be good to have on record — not a merge condition.

Merge coordination (restating, since it affects sequencing): this touches the same three places as #4806 and #4571 — a field on RecoverableConnection, its constructor, and the tail of mod tests in connection.rs — all off base 97e0d63c. All three will conflict. This one is the most orthogonal, so it's cheapest to land last.