問題

ユーザー管理ではユーザーを識別する値は重複することは通常許されない。例えばメールアドレスをユーザー識別に使う場合、同じアドレスが登録できないことを保証する必要がある。

SQLデータベースならば以下のようにusersテーブルのemailカラムにunique制約をつければ良い。

CREATE TABLE IF NOT EXISTS "users" (
  "id" integer PRIMARY KEY AUTOINCREMENT NOT NULL,
  "name" varchar,
  "email" varchar,
   ...
);

CREATE UNIQUE INDEX "index_users_on_email" ON "users" ("email");

しかしDatastoreではプロパティ（SQLのカラムに相当）にunique制約をかけることができないので同様のことをするには工夫がいる。

案1（マルチスレッドに慣れていないとやりがちな抜け穴のある案）

すぐに思いつくのはメールアドレスが既に登録されているかどうかを確認し、登録されていればエラーとする方法である。Pythonで書くと以下の様な感じ。ユーザー登録に成功すればそのエンティティを返し、失敗すればNoneを返す。

from google.cloud import datastore

KIND_USERS = 'users'

def register_user(name, email):
    client = datastore.Client()

    # emailアドレスが登録済みかどうかを調べる
    query = client.query(kind=KIND_USERS)
    query.add_filter('email', '=', email)
    users = list(query.fetch())
    if 0 < len(users):
        # 既にemailアドレスは登録済み
        return None

    # ユーザー登録を行う
    key = client.key(KIND_USERS)
    user = datastore.Entity(key)
    user['name'] = name
    user['email'] = email
    client.put(user)
    return user

しかしこの案は以下のケースで失敗する。

Aさんがメールアドレス foo@bar.com で登録を開始する。
Aさんはこのメールアドレスがまだ未登録であることを確認する。
Bさんが同じメールアドレスで登録を開始する。
何らかの理由で時間がかかり、まだAさんは登録が済んでないためBさんもこのメールアドレスがまだ未登録であることを確認する。
AさんもBさんもユーザー登録を行う。

これは以下のテストコードで確認できる。シナリオとしては

Aさんが登録を開始しメールアドレスが登録されていないことを確認した後、ユーザー登録するまで何らかの理由で2秒ほど時間がかかっていると想定する。
一方 Bさんは1秒ほど後から登録を開始し、すぐに登録を終了すると想定する。

from google.cloud import datastore
import time
import threading

KIND_USERS = 'users'

def register_user(name, email, start_delay, busy_period):
    # 処理開始を調整
    time.sleep(start_delay)
    print(f'{name}: 処理開始')

    client = datastore.Client()

    # emailアドレスが登録済みかどうかを調べる
    query = client.query(kind=KIND_USERS)
    query.add_filter('email', '=', email)
    users = list(query.fetch())
    print(f'{name}: {email}が登録済みかを確認')
    if 0 < len(users):
        # 既にemailアドレスは登録済み
        print(f'{name}さん、{email}は既に登録済みです')
        return None

    # 何らかの理由で処理が遅延していることを模擬している
    time.sleep(busy_period)

    # ユーザー登録を行う
    key = client.key('users')
    user = datastore.Entity(key)
    user['name'] = name
    user['email'] = email
    client.put(user)
    print(f'{name}: 登録終了')
    return user

def show_all_users():
    client = datastore.Client()
    query = client.query(kind=KIND_USERS)
    users = list(query.fetch())
    print('** 登録ユーザー一覧 **')
    for user in users:
        print(f'{user["name"]}: {user["email"]}')

def delete_all_users():
    client = datastore.Client()
    query = client.query(kind=KIND_USERS)
    users = list(query.fetch())
    for user in users:
        key = client.key(KIND_USERS, user.key.id)
        client.delete(key)

if __name__ == '__main__':
    email = 'foo@bar.com'
    # 念の為前回の登録を最初に消す
    delete_all_users()

    # AとBが同時に同じemailアドレスを登録すると想定。
    #
    # Aはすぐに処理を開始しemailアドレスが登録されていないことを
    # 確認した後、ユーザー登録するまで何らかの理由で2秒ほど時間が
    # かかっていると想定。
    #
    # 一方 Bは1秒ほど後から処理を開始し、すぐに登録を終了すると想定。
    t1 = threading.Thread(target=register_user, args=('A',email,0,2))
    t2 = threading.Thread(target=register_user, args=('B',email,1,0))
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    # 登録ユーザーを確認
    show_all_users()
    # 登録を削除
    delete_all_users()

結果は以下のようになり、同じメールアドレスが登録されてしまっている。

A: 処理開始
A: foo@bar.comが登録済みかを確認
B: 処理開始
B: foo@bar.comが登録済みかを確認
B: 登録終了
A: 登録終了
** 登録ユーザー一覧 **
B: foo@bar.com
A: foo@bar.com

この対策としては以下が考えられる。

Datastoreを使うのは止めて、SQLデータベースを使用する。

ごもっとも。でもDatastoreならある程度なら無料なんです。

同じメールアドレス（やアカウント名）を同時に登録するなんて確率的にありえないので、上記のコードで良しとする。

仲間内で使うような小さなWebアプリならそれでも良いかも。例えばログイン時にメールアドレスが複数登録されていたらエラーとしてログインできないようにする。そしてユーザーに報告してもらうかエラーログを見て対処する。でもちゃんと重複を禁止するのはそれほど手間ではないのでやっておいた方が吉だと思う。この辺りで手を抜くといつかセキュリティがらみで手痛いしっぺ返しを食らいそう…

メールアドレス用のkindを作成し、メールアドレスをエンティティの識別子とする。さらにDatastoreのトランザクション機能を使って重複禁止を実現する。

以下はその説明。

案2（途中で手が止まった案）

Datastoreドキュメントの「エンティ、プロパティ、キー」にあるようにキーは各エンティティで一意である。よってメールアドレス用のカインドを作成し、メールアドレスをキーの識別子にすれば重複を防げる。

SQLデータベースから類推すると、メールアドレスをキーとしたエンティティを作成し、SQLデータベースの insert に相当することをすれば既に登録されている場合はエラーとなるはずである。

しかしエンティティを保存する client.put(entity) はキーがまだ存在していなければ insert 相当の動作になるが、存在していれば update 相当の動作になり、重複しているかどうかにかかわらず成功する。

from google.cloud import datastore

KIND_EMAILS = 'emails'

def check_and_register_email(email, n):
    client = datastore.Client()
    # emailを識別子としてキーを作成
    key = client.key(KIND_EMAILS, email)
    # エンティティを作成
    entity = datastore.Entity(key)
    # SQLデータベースの insert に相当することをしたいが
    # 以下はkeyが存在していなければ insert
    # 存在していれば update の動作になってしまい、
    # 重複しているかどうかにかかわらず成功する
    client.put(entity)
    print(f'{n}回目の登録成功')

def show_all_emails():
    client = datastore.Client()
    query = client.query(kind=KIND_EMAILS)
    emails = list(query.fetch())
    print('** 登録メールアドレス一覧 **')
    for email in emails:
        print(f'email: {email.key.name}')

def delete_all_emails():
    client = datastore.Client()
    query = client.query(kind=KIND_EMAILS)
    emails = list(query.fetch())
    for email in emails:
        key = client.key(KIND_EMAILS, email.key.name)
        client.delete(key)

if __name__ == '__main__':
    email = 'foo@bar.com'
    # 念の為前回の登録を最初に消す
    delete_all_emails()
    # 最初の登録
    check_and_register_email(email, 1)
    # 2回目
    check_and_register_email(email, 2)
    # 登録メールアドレスを表示
    show_all_emails()
    # 登録を削除
    delete_all_emails()

以下が実行結果。

1回目の登録成功
2回目の登録成功
** 登録メールアドレス一覧 **
email: foo@bar.com

このように insert 相当の動作をさせ重複時に失敗させたいができないことになる。

案3（やっと成功した案）

そうするとメールアドレスを識別子としたエンティティが既に登録されているかを確認し、されていたらエラーとする必要がある。

最初の案に戻ってしまったようであるが、 Datastoreのトランザクションを使うことでどのようなタイミングでも重複時に失敗させることができる。

分離と整合性の節にあるようにトランザクション内では同じエンティティを同時に変更することはできない。

同時に変更が起こった場合には１つのみが成功し他は失敗する。

マルチスレッドプログラミングでの排他制御と似ているが wait しないで実行し最後にトランザクション失敗になる点が異なる。

これは以下のテストコードで確認できる。シナリオとしては

Aさんがメールアドレス登録を開始し登録されていないことを確認した後、登録するまで何らかの理由で2秒ほど時間がかかっていると想定する。
一方 Bさんは1秒ほど後から登録を開始し、すぐに登録を終了すると想定する。

from google.cloud import datastore
import time
import threading

KIND_EMAILS = 'emails'

def check_and_register_email(name, email,
                     start_delay=0, busy_period=0):
    # 処理開始を調整
    time.sleep(start_delay)
    print(f'{name}: 処理開始')

    client = datastore.Client()

    # emailを識別子としてキーを作成
    key = client.key(KIND_EMAILS, email)

    try:
        with client.transaction():
            print(f'{name}: トランザクション開始')
            # emailを識別子としたentityを取得
            entity = client.get(key)
            print(f'{name}: {email}が登録済みかを確認')
            if entity:
                # 既にemailアドレスは登録済み
                print(f'{name}さん、{email}は既に登録済みです')
                return None

            # 何らかの理由で処理が遅延していることを模擬している
            time.sleep(busy_period)

            # emailを識別子としたentityを新たに作成して登録する
            entity = datastore.Entity(key=key)
            # 何でも良いので１つ以上プロパティを作って値を入れる。
            # Entityはdict（辞書）から派生しているのでそうしないと
            # 上の if entity: が常にFalse になる。
            entity['name'] = name
            print(f'{name}: 登録開始')
            client.put(entity)
    except Exception as e:
        # 同じエンティティを登録しようとしたためtransactionの競合が起きて
        # 例外が発生した
        print(f'{name}: 例外発生: {type(e)} {e}')
        print(f'{name}: 登録失敗')
        return None

    print(f'{name}: 登録成功')
    return entity

def show_all_emails():
    client = datastore.Client()
    query = client.query(kind=KIND_EMAILS)
    emails = list(query.fetch())
    print('** 登録メールアドレス一覧 **')
    for email in emails:
        print(f'{email["name"]}: {email.key.name}')

def delete_all_emails():
    client = datastore.Client()
    query = client.query(kind=KIND_EMAILS)
    emails = list(query.fetch())
    for email in emails:
        key = client.key(KIND_EMAILS, email.key.name)
        client.delete(key)

if __name__ == '__main__':
    email = 'foo@bar.com'
    # 念の為前回の登録を最初に消す
    delete_all_emails()

    # AとBが同時に同じemailアドレスを登録すると想定。
    #
    # Aはすぐに処理を開始しemailアドレスが登録されていないことを
    # 確認した後、ユーザー登録するまで何らかの理由で2秒ほど時間が
    # かかっていると想定。
    #
    # 一方 Bは1秒ほど後から処理を開始し、すぐに登録を終了すると想定。
    t1 = threading.Thread(target=check_and_register_email, args=('A',email,0,2))
    t2 = threading.Thread(target=check_and_register_email, args=('B',email,1,0))
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    # どちらが登録されたかを確認
    show_all_emails()
    # 登録を削除
    delete_all_emails()

結果は以下のようになり1人のみが登録に成功し、他は失敗している。

A: 処理開始
A: トランザクション開始
A: foo@bar.comが登録済みかを確認
B: 処理開始
B: トランザクション開始
B: foo@bar.comが登録済みかを確認
B: 登録開始
A: 登録開始
B: 例外発生: <class 'google.api_core.exceptions.Aborted'> 409 Aborted due to cross-transaction contention. This occurs when multiple transactions attempt to access the same data, requiring Firestore to abort at least one in order to enforce serializability.
B: 登録失敗
A: 登録成功
** 登録メールアドレス一覧 **
A: foo@bar.com

なお注意点としては、メールアドレス用のエンティティには何でも良いので１つプロパティを作ること。Datastore の Entity クラスは dict から派生しており、プロパティが１つもないと空の dict となり if entity: の判断が常に False になる。上記の例ではnameというプロパティを作成している（うっかり要らないからとコメントアウトしてハマりました）。又は if entity: の部分を if entity is not None: と直接 None と比較しても良い。