Web アプリケーションにおける排他制御の実装例 (第一回)

ステータスについては、enum で特定の値のみが入力されるように定義しています。実案件では他の要件もあったことからもう少しステータスの種類が多かったのですが、説明のために簡略化しています。

• normal：通常状態
• editing：編集状態
• locked：ロック状態（＝外部キー参照元のテーブルが編集中）

また、ステータスの遷移については、以下の通り通常状態をベースに編集状態またはロック状態に遷移するような定義とします。（ちゃんとした状態遷移図でないのはご了承ください・・） 

Lambda における設計・実装

DynamoDB のテーブルに対する汎用オペレーション（get, list, create など）については Amplify が自動生成してくれる graphql の query / mutation を使用していますが、残念ながら「条件付き書き込み」や「TransactWriteItems」を実現してくれるようなスキーマ定義の方法はありません。よって、テーブルステータスの更新については、Lambda で実装したものを AppSync 経由で実行するような棲み分けとしています。

Amplify（schema.graphql）における mutation のスキーマ定義例

type を input として使い回せないのが唯一イマイチな仕様だなと思っているのですが、それ以外は特に違和感なく使えています。機能・記法としてこういうものが欲しいみたいな話はまた別にありますが。。

import os
import boto3
import traceback
from datetime import datetime, timezone, timedelta
from botocore.exceptions import ClientError

client = boto3.client('dynamodb')

def lambda_handler(event, context):
    try:
        # 引数、返り値の初期化
        result_json = {}
        result_json['lock_result'] = False

        table_name = {'S': event['arguments']['input']['table_name']}
        current_status = {'S': event['arguments']['input']['current_status']}
        target_status = {'S': event['arguments']['input']['target_status']}
        fk_table_name = [{'S': x} for x in event['arguments']['input']['fk_table_name']] if 'fk_table_name' in event['arguments']['input'] else []
        fk_current_status = {'S': event['arguments']['input']['fk_current_status']} if 'fk_current_status' in event['arguments']['input'] else None
        fk_target_status = {'S': event['arguments']['input']['fk_target_status']} if 'fk_target_status' in event['arguments']['input'] else None
        editor = {'S': event['arguments']['input']['editor']} if 'editor' in event['arguments']['input'] else {'NULL': True}
        locked_by = {'S': event['arguments']['input']['table_name']} if 'locked_by' in event['arguments']['input'] else {'NULL': True}
        current_time = {'S': datetime.now(timezone.utc).isoformat(timespec='milliseconds').replace('+00:00','Z')}

        # FK参照先テーブルロック解除時、editorを削除するように設定
        if event['arguments']['input']['target_status'] == 'normal':
            fk_editor = {'NULL': True}
        else:
            fk_editor = editor

        # 対象テーブルの状態遷移
        target_items = [{
            'Update': {
                'TableName' : <MASTER_TABLE_NAME>,
                'Key' : {
                    'name': table_name,
                },
                'UpdateExpression' : "SET #st = :upd_val1, updatedAt = :upd_val2, editor = :upd_val3",
                'ConditionExpression' : "#st = :cond_val1",
                'ExpressionAttributeNames' : {'#st' : 'status'},
                'ExpressionAttributeValues' : {
                    ':upd_val1': target_status,
                    ':upd_val2': current_time,
                    ':upd_val3': editor,
                    ':cond_val1': current_status
                }
            }
        }]

        # 対象テーブルのFK参照先テーブルの状態遷移
        for lock_table in fk_table_name:
            target_items.append({
                'Update': {
                    'TableName': <MASTER_TABLE_NAME>,
                    'Key': {
                        'name': lock_table,
                    },
                    'UpdateExpression': "SET #st = :upd_val1, updatedAt = :upd_val2, editor = :upd_val3, locked_by = :upd_val4",
                    'ConditionExpression': "#st = :cond_val1",
                    'ExpressionAttributeNames': {'#st' : 'status'},
                    'ExpressionAttributeValues': {
                        ':upd_val1': fk_target_status,
                        ':upd_val2': current_time,
                        ':upd_val3': fk_editor,
                        ':upd_val4': locked_by,
                        ':cond_val1': fk_current_status
                    }
                }
            })

        # 条件付き書き込みの実施
        update_response = client.transact_write_items(
            TransactItems=target_items
        )

        # 返り値の設定
        result_json['lock_result'] = True
        result_json['current_status'] = event['arguments']['input']['target_status']
        result_json['fk_current_status'] = []
        for lock_table in fk_table_name:
            result_json['fk_current_status'].append({'table_name': lock_table['S'], 'current_status': fk_target_status['S']})

    # 条件付き書き込み失敗時の処理
    except ClientError as e:
        print(traceback.format_exc())
        try:
            # 返り値の設定
            response = client.get_item(
                TableName = <MASTER_TABLE_NAME>,
                Key={'name': {'S': event['arguments']['input']['table_name']}}
            )
            result_json['current_status'] = response['Item']['status']['S']
            result_json['fk_current_status'] = []
            for lock_table in fk_table_name:
                response = client.get_item(
                    TableName = <MASTER_TABLE_NAME>,
                    Key={'name': lock_table}
                )
                result_json['fk_current_status'].append({'table_name': lock_table['S'], 'current_status': response['Item']['status']['S']})

        except Exception as e:
            print(traceback.format_exc())
   
    finally:
        return result_json

実装例の内容について全て説明すると長くなってしまうので、要点のみ箇条書きでまとめます。

• 冒頭で、AppSync から渡される引数と、AppSync に返す返り値の初期化をしています。
• TransactWriteItems により、単一のトランザクション内で実行したいクエリを target_items リストに格納しています。
  • 外部キー（FK）のないテーブルは、TransactWriteItems で実行するクエリ数は1つです。
  • 外部キー（FK）のあるテーブルは、外部キー参照先テーブルの個数分クエリ数が増加します。
• 各クエリはステータス更新を伴うため、全て条件付き書き込みを使用しています。
  • ‘ConditionExpression’ で条件付き書き込みの「条件」を指定しています。今回の条件は、ステータスが想定通りの値となっているかとなります。
  • ‘ExpressionAttributeNames’ は、DynamoDB スキーマ定義で使用している「status」という単語が予約語にあたるようで、そのまま使用するとエラーになってしまうため、クエリ内で一時的に別の単語に置き換える目的で使用しています。
    • https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/ReservedWords.html
  • つまり、TransactWriteItems で実行される各クエリの内1つでも失敗した場合は、トランザクション内の他のクエリについても実行されず、実質的に対象テーブルのロックを取得できなかったことになります。
• 条件付き書き込みが失敗した場合（≒対象テーブルのロックを取得できなかった場合）は ClientError が発生するので、それを except 句で拾って例外処理を実装しています。

後は、この Lambda 関数を AppSync API 経由で実行するような画面を作れば、ユーザがアプリケーション上でテーブルのステータスを変更する際に排他制御御（ロック）が実現できるということになります。