fix: make async loop creation thread-safe

[Carter Tinney (cartertinney)]

Summary

• serialize first-time creation of the three shared async client event loops
• preserve lock-free access after a loop has been published
• synchronize test-only loop cleanup with loop creation
• add concurrent initialization and Janus loop-affinity regression coverage

Root cause

Python E2E build 163209 timed out in the first async C2D test on Windows/Python 3.10. Message delivery succeeded, but the log showed CLIENT_INTERNAL_LOOP being created twice by concurrent first callers.

The loop getters used an unsynchronized check-then-create. After #1258 stopped constructing each Janus queue through the internal loop, the loop was no longer guaranteed to be initialized before the handler runner and feature-enablement paths started concurrently. The queue could bind to the first loop while the second loop replaced the global reference. Every subsequent queue receive then failed Janus's loop-affinity check, and the handler manager's immediate restart behavior turned the failure into a tight restart loop until pytest timed out.

Why this shape

The process-wide loops are intentionally shared by clients to isolate internal work, handler runners, and user coroutine handlers. Moving to per-client loops would require broader ownership and shutdown changes while leaving lazy initialization subject to the same race. Eager creation would unconditionally start three daemon threads. A centralized, double-checked creation lock restores the existing singleton invariant at its owner and keeps the established hot path lock-free.

Why prior review missed it

#1258's reviews and tests focused on Janus resource closure and removing an unnecessary event-loop round trip during queue construction. Its unit tests verified sequential construction, queue operations, and shutdown, and its complete E2E matrix passed. The older loop manager only asserted that repeated sequential calls returned the same loop; it had no concurrent-first-access coverage.

The change therefore exposed a pre-existing race rather than introducing an obviously incorrect Janus operation. It requires two independent first consumers to enter a narrow scheduling window, which did not occur in #1258's runs and appeared later in one Windows job. This PR makes that concurrency contract explicit and deterministic in tests.

Validation

• regression demonstrated before the fix: all three loop getters created two loops under synchronized concurrent first access
• Python 3.10 focused async lifecycle suite: 153 passed
• Python 3.14 related async lifecycle suite: 933 passed, 3 skipped
• full unit suite: 5445 passed, 6 skipped
• concurrency regression repeated 20 times
• package sdist and wheel build succeeded
• Black and Ruff passed
• independent design and code reviews completed

[Copilot]

🟡 Changes recommended

The concurrency regression test relies on timing and can false-pass against the original race.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Adds synchronized lazy initialization for shared asynchronous client event loops.

Changes:

• Protects loop creation and cleanup with a shared lock.
• Adds concurrent initialization regression coverage.
• Verifies Janus queue loop affinity.

File summaries

File	Description
loop_management.py	Serializes event-loop creation and cleanup.
test_loop_management.py	Tests concurrent first access.
test_async_inbox.py	Tests Janus loop affinity.

Review details

• Files reviewed: 3/3 changed files
• Comments generated: 1
• Review effort level: Balanced

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

🟢 Approval recommended

The thread-safety fix and regression coverage are complete, with no unresolved issues.

Review details

• Files reviewed: 3/3 changed files
• Comments generated: 0 new
• Review effort level: Balanced

[Carter Tinney (cartertinney)]

/azp run Azure.azure-iot-sdk-python-dps-e2e

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).