Concurrently creating firebase users of the same email succeeds

The reproduction is outlined in firebase_service.py and test.py; start by running the ThreadPoolExecutor case against Firebase Authentication and compare it with the asyncio case. Investigate the SDK path used by auth.create_user and determine whether the concurrent requests are handled consistently. Done means the concurrent case no longer creates multiple users with the same email and the expected EmailAlreadyExistsError behavior is verified.

[READ] Step 1: Are you in the right place?

• For issues related to the code in this repository file a GitHub issue.
• If the issue pertains to Cloud Firestore, report directly in the
  Python Firestore GitHub repo. Firestore
  bugs reported in this repo will be closed with a reference to the Python Firestore
  project.
• For general technical questions, post a question on StackOverflow
  with the firebase tag.
• For general Firebase discussion, use the firebase-talk
  google group.
• For help troubleshooting your application that does not fall under one
  of the above categories, reach out to the personalized
  Firebase support channel.

[REQUIRED] Step 2: Describe your environment

• Operating System version: MacOS sonoma 14.3
• Firebase SDK version: 6.5.0
• Firebase Product: auth
• Python version: 3.11
• Pip version: 23.1.2

Note that I have used other tenants on Google Identity Platform in this gcp project but I scrapped that and went back to the single tenant.
This issue is happening on the default firebase tenant. (Sharing this fact in case it somehow affects the outcome)

[REQUIRED] Step 3: Describe the problem

I discovered that my tests for some CRUD functionality were creating duplicate firebase users with the same email which is problematic because we rely on the firebase_admin._auth_utils.EmailAlreadyExistsError to safeguard functionality. This appeared as my webserver API was called to create the same user concurrently and it actually made multiple firebase users.

For context, my firebase project -> authentication -> settings -> user account linking -> link accounts with same email is active.

Steps to reproduce:

Creating a new user with an existing email address succeeds when its ran concurrently. I have tested 3 scenarios:

1. Create the new user (with existing email) a second after the existing user was created
   1.1. I get the firebase_admin._auth_utils.EmailAlreadyExistsError like I expect

2. Attempt to create 4 users with the same email address using asyncio library to create them quickly
   2.1 Returns one created user and fires 3 firebase_admin._auth_utils.EmailAlreadyExistsError like I expect

3. Attempt to create 4 users with the same email address concurrently using ThreadPoolExecutor
3.1 Returns 4 new Firebase users who all share the same email address. Not good.

Calling the auth.get_user_by_email(email) returns the latest firebase user created, when I delete the latest one then the function returns the newest one after that and so on.

I plan to add in concurrency/idempotent protections to my API logic in the mean time as this would cause a mess downstream ( as uncommon as it would occur)

Relevant Code:

This is a rough outline of my tests but in general just call create_user concurrently to hopefully see the same results. I use a Fastapi server so you can ignore some of this extracted code using async where its not needed. This is just to demo the issue.
firebase_service.py

import firebase_admin
from firebase_admin import auth
if not firebase_admin._apps:
    firebase_admin.initialize_app()

class FirebaseService:

  # async wrapper
  async def create_user(self, email: str, uid: str | None) -> auth.UserRecord | None:
          """Create a user."""
          try:
              user = auth.create_user(email=email, uid=uid)
              return user
          except Exception as e: 
              logger.exception(f"error creating user {email} : {e}")
              return None
  
  # non async implementation	
      def create_user_sync(self, email: str, uid: str | None) -> auth.UserRecord | None:
          """Create a user."""
          try:
              user = auth.create_user(email=email, uid=uid)
              return user
          except Exception as e: 
              logger.exception(f"error creating user {email} : {e}")
              return None

test.py

import asyncio

common_email = "email_goes_here"
no_uid = None

async def test_multiple_user_same_email_create_asyncio() -> None:
    """Test multiple user creation with same email with asyncio gather."""
    tasks = []
    for i in range(4):
        tasks.append(asyncio.create_task(firebase_service.create_user(uid=no_uid, email=common_email)))
    results = await asyncio.gather(*tasks)
    logger.info(results) # correctly creates 1 user and raises an error for the rest

async def test_multiple_user_same_email_create_threadpool() -> None:
    """Test multiple user creation with same email concurrently"""
    uid = None

    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(firebase_service.create_user_sync,common_email, no_uid ) for _ in range(4)]
        results = [future.result() for future in as_completed(futures)]
        logger.info(results) # actually creates 4 users with the same email

if __name__ == "__main__":
    # asyncio.run(test_multiple_user_same_email_create_asyncio())
     # asyncio.run(test_multiple_user_same_email_create_threadpool())