なぜ二重書き込みが問題になるのか

注文状態のデータベース（DB）更新と、返金タスクの送信を別々に行うと、片方だけ成功する可能性があります。この不整合が起こり得る時間帯を、障害窓と呼びます。DB更新後に送信が失敗すれば、承認済みなのに返金タスクがありません。送信後にDB更新が失敗すれば、承認前の注文に返金処理が走るおそれがあります。[1, 6]

Spanner queuesは、注文状態の更新とキューへのタスク投入を同じSpannerトランザクションに入れられます。トランザクションは一連のDB操作をまとめて確定する単位です。トランザクションがコミット（変更の確定）されれば、両方が確定します。トランザクションが確実に中止されれば、両方とも反映されません。この全体が一緒に確定する性質を、原子性と呼びます。[2, 5]

応答のタイムアウトだけでは、トランザクションの中止を判断できません。確定後に応答だけ失われるunknown commit（確定結果が不明な状態）が起きることもあります。状態照合は、この不確実性に対処するための設計上の考察です。[20]

外部MQ・outbox・change streamsとの違い

外部MQは、アプリケーションから独立したメッセージキューです。Transactional outbox（業務更新とイベント記録を同じDBトランザクションに保存する方式）は、二重書き込みを避けながら外部MQへ届ける方法です。ただし、別プロセスがイベント行を読み、外部MQへ送る段階は元のトランザクションの外にあります。[6, 7]

Change streams（DB変更を読み出す仕組み）は、データ変更と変更記録を同じトランザクションに保存します。記録を読み出して外部サービスへ届ける処理は、別のストリーム読取段階です。[8]

Spanner queuesは、Spanner内の状態更新とメッセージ送信を同じトランザクションにまとめられます。受信側では、業務更新とACK（処理済みとしてキューから削除する確認）を同じトランザクションにまとめられます。外部システムへの配送が必要なら、outboxやchange streamsが引き続き選択肢になります。[2, 8, 16]

方式DB更新と記録・投入の原子性外部MQへの配送
外部MQへの直接二重書き込みDB更新と送信は別々ですアプリケーションが直接送信します
Transactional outboxDB更新とイベント行を同時に確定します別プロセスが外部MQへ転送します
Change streamsDB更新と変更記録を同時に確定しますストリーム読取段階で外部へ届けます
Spanner queues送信側は業務更新＋投入、受信側は業務更新＋ACK（処理済みとしてキューから削除する確認）を、それぞれのトランザクションで確定しますSpanner内のキューを消費します。外部MQへの転送は別処理です

表の原子性（すべて成功するか、すべて失敗する性質）はDB内でどこまで一緒に確定できるかを示します。outboxとchange streamsでは外部への配送が後続段階です。Spanner queuesも、外部MQや外部API（外部サービスの機能を呼び出す窓口）まで一つのトランザクションに含める機能ではありません。[6, 7, 8, 2, 4]

注文返金で見る処理経路と障害窓

返金承認の処理では、注文状態を更新し、返金タスクを投入します。次のコードは、SQL（DBを操作する言語）のうち、SpannerのGoogleSQL方言で書いた例です。

OrdersにはOrderId、Status、RefundTransactionIdがあるとします。RefundTasksの主キーはOrderIdとTaskIdです。この複合主キーは、複数列を組み合わせて行を識別します。PayloadはBYTES型（バイト列）として定義済みとします。DeliverTimeは配信時刻の列として例で使用します。[5, 9]

以降は、トランザクションをTXと略します。公式資料が示すrunner型API（TX関数を実行し、TXの確定や中止を管理するAPI）の例は、GoクライアントライブラリのReadWriteTransactionです。TX関数からエラーを伝播させると、ライブラリがTXを破棄してロールバックします。ここではこれをrunnerと呼び、以下の2文を同じ読み書きTXで実行する前提とします。[18]

返金待ち状態はREFUND_PENDINGとします。設計上、SDK（ソフトウェア開発キット）側で注文の更新件数が1件であることを確認します。更新件数が0件なら、アプリケーションはTX関数からエラーをrunnerへ返します。runnerはTX全体をロールバックするため、タスクは投入されません。

UPDATE Orders
SET Status = 'REFUND_APPROVED'
WHERE OrderId = @orderId AND Status = 'REFUND_PENDING';
-- SDK側で更新件数1を確認。違えばTX関数からエラーをrunnerに返し、TXを中止。

INSERT INTO RefundTasks (OrderId, TaskId, Payload, DeliverTime)
VALUES (@orderId, @taskId, @payload, CURRENT_TIMESTAMP());

受信はpull型です。タスクを受信して処理するプログラムであるワーカーが、キューにメッセージを要求して受け取る方式です。RECEIVE_RefundTasks()をExecuteStreamingSQLで実行します。ストリーミング呼び出し（結果を順次受け取る呼び出し）は長時間実行できます。呼び出しの存続時間とメッセージごとの初期lease（ワーカーに一時的な処理権を与える期間）は別です。SDKでの具体的な読み取り条件は、公式の利用手順を参照してください。[9]

SELECT OrderId, TaskId, Payload, DeliverTime,
       SpannerLeaseExpirationTimestamp, SpannerLeaseToken
FROM RECEIVE_RefundTasks();

受信時の初期leaseは10秒です。leaseは、メッセージを処理するワーカーに与えられる一時的な処理権です。処理が長引く場合は、RENEWLEASE_RefundTasks()で更新します。[10]

更新が成功すると、新しいlease token（leaseを識別し、更新に使う値）と有効期限が返ります。ワーカーはその値を保存し、次の更新には新しいtokenを使います。ワーカーがleaseの期限前にleaseを更新しなければ、メッセージは再配信され得ます。[10]

SELECT *
FROM RENEWLEASE_RefundTasks(lease_tokens => [@leaseToken]);

ACK（処理済みとしてキューから削除する確認）されないままleaseが切れると、メッセージは再配信されます。古いワーカーが処理を続けている間に別のワーカーが同じタスクを受け取ることもあります。Googleは、10秒を超える処理ではleaseを定期更新し、処理開始から約7〜8秒で更新する方法を案内しています。[11, 12, 13]

返金APIはSpannerのトランザクション外で呼びます。ここでは設計上の選択肢として、ワーカーはAPIの成功を確認するか、API提供者が結果照会手段を提供する場合は照合した上で、別のSpannerトランザクションで結果を記録し、ACKします。照会できない場合はAPIの仕様に応じて処理します。外部APIとACKは原子的に結び付けられないため、API成功後・ACK前にワーカーが停止すれば、再配信後にAPIが再度呼ばれ得ます。[21, 22]

APIの冪等性キー（同じ要求の重複実行を抑える識別子）は、提供者が定める保存・照合範囲で同一要求の重複適用を抑えます。Spannerのコミットとの原子性や、無期限の重複防止まで保証するものではありません。[23, 24]

後輩DBで二重処理を防げても、返金APIの結果が分からない時間は残りますね。設計では何を基準に分けるんですか？

先輩DB内の更新とACKは一つにまとめ、外部APIは別の境界として扱うことだね。API側が冪等性キー（同じ要求の重複実行を抑える識別子）に対応するなら、同じ要求の再試行を重複適用しないための手掛かりになるよ。

後輩そのキーがあれば、Spannerのコミットまで一度だけにできるんでしょうか？

先輩そこまでは保証しないよ。キーの効力はAPI提供者が定める保存・照合範囲に限られる。DBのコミットとAPIの結果を一体化するものではないんだ。

「一度だけ」が成立する範囲

Spanner queuesの配信は少なくとも一度で、同じメッセージが再処理される可能性があります。一方、1つのメッセージに対するACKが成功するのは最大1回です。ACKが一度だけという性質だけでは、業務処理も一度だけになるとは限りません。[3]

DB内の重複更新を抑えるには、メッセージに関係する業務更新とACKを同じSpannerトランザクションに入れます。さらに、ACKのDELETEにASSERT_ROWS_MODIFIED 1を付け、対象メッセージを1行削除できた場合だけACKを成功させます。次は、結果更新と複合主キーを指定したACKを、同じ読み書きTXで実行する構文例です。ここでも、アプリケーションは注文の更新件数を確認し、対象注文がなければTX全体をロールバックする設計にします。[16]

UPDATE Orders
SET RefundTransactionId = @refundTransactionId
WHERE OrderId = @orderId;

DELETE FROM RefundTasks
WHERE OrderId = @orderId AND TaskId = @taskId
ASSERT_ROWS_MODIFIED 1;

ASSERT_ROWS_MODIFIED 1が失敗すると、OUT_OF_RANGE（指定した更新件数と実際の件数が一致しない場合のエラー）が返ります。ただし、これは文レベルのエラーです。Spannerがトランザクション全体を自動で中止したり、先行する書き込みをロールバックしたりするわけではありません。公式資料では、GoのReadWriteTransactionなどのrunner型APIを使う場合、TX関数からエラーを伝播させるよう案内しています。手動でTXを管理する場合は、明示的にロールバックする必要があります。[17, 18]

エラーを捕捉して処理を続けると、DELETEだけ失敗し、同じトランザクション内の業務更新がコミットされ得ます。これでは重複を防ぐための条件が崩れます。再試行時に対象行がすでにない場合も同じエラーになるため、エラーを捕捉して処理を続ける実装は避けます。[14, 15, 19]

コミット要求後に応答が失われると、トランザクションが確定したか分からないunknown commitが起こり得ます。[20]

注文状態やタスクの状態を照合し、APIが対応する場合は同じ冪等性キー（同一要求の重複実行を抑える識別子）と同じパラメーターで再試行する、という手順は設計上の考察です。冪等性キーの保持期間や照合条件は、API提供者の仕様を個別に確認します。例えばStripeの仕様を他の返金APIへ一般化することはできません。[20, 23]

運用と導入判断

Spanner queuesはEnterpriseとEnterprise Plusエディションで利用できます。Spanner queuesは、SpannerのGoogleSQL方言と、PostgreSQL（オープンソースのDB製品名）のSQL方言に対応します。[25, 26]

1キューにつき、同じ引数で同時実行できる受信クエリは最大1,000件です。受信TVF（表形式の結果を返す関数）の同時実行上限は、プロジェクト・リージョン単位で2,000件です。キュー数は100処理ユニット（Spannerの容量を表す単位）あたり10キュー、インスタンスあたり最大100キューです。[25, 26, 27, 28]

キューの送受信やDB内の更新はSpannerのコンピュート資源（DB処理に使う計算能力）を使います。ワーカーの実行基盤によっては、Spannerとは別に計算資源を見積もります。Spannerのコンピュート資源を共有するため、既存DBの処理負荷と競合する可能性があります。導入前には、エディション、同時受信数、キュー数、コンピュート容量を確認します。性能や費用が外部MQより優れるとは、ここで示された仕様だけからは判断できません。[29]

Cloud Monitoring（Google Cloudの監視サービス）では、最古の未ACKメッセージの年齢、送信数、ACK数、lease失効数などを確認できます。これらを組み合わせて、処理の滞留と失敗を切り分けます。[31]

未ACKメッセージにはTTL（有効期間に基づく削除）を設定できます。TTLは期限到来後すぐの削除を保証しません。Spannerは対象データを通常72時間以内に削除します。72時間は必ず削除される期限ではありません。未処理タスクが削除対象になり得るため、保持期間と滞留状況を合わせて設計します。[30]

失敗または未ACKのメッセージは間隔を広げるバックオフ付きで再試行され、最初の1時間を過ぎると1時間ごとに再試行されます。Googleは、恒久的に失敗するメッセージを別キューへ移すことを検討するよう案内しています。遅延タスクを取り消す場合、キューのメッセージ削除だけで、すでに始まった外部返金を取り消せるわけではありません。これは外部APIとSpannerのACKが別の処理であることから導かれる設計上の注意点です。[32, 4]

Spanner内の状態変更と非同期タスクを同時に確定したい処理では、queuesが候補になります。外部MQへの配送や複数システムへの連携が必要なら、outboxやchange streamsを含めて配送境界を選びます。どの方式でも、外部APIの副作用（返金など、外部側に生じる状態変更）には、DB内のACKとは別の冪等性設計が必要です。[2, 7, 8, 4]