Spanner Change Streams を自前で読む ― 決済データを欠損なく取り込む

はじめに

こんにちは、solaris チームでバックエンドを開発している大迫(m11o)です。

私たちが開発している solaris は、UPSIDER の法人カード基盤の一部を担うサービスです。UPSIDER では、自社で磨いてきた法人カード基盤を「パートナー企業が自社ブランドのカード事業を立ち上げられる形」（マルチブランド対応）で外部提供する取り組みを進めており、その基盤は複数のマイクロサービスで構成されています。solaris はその中でカード管理基盤、つまりカードの発行・利用枠の管理や VISA 決済処理まわりを担当しています。

この全体像（マルチブランド対応や、なぜマイクロサービスに分割したのか）については、すでに公開されている以下の記事がとても詳しいので、そちらに譲ります。

この記事でお話しするのは、もっと狭くて具体的なテーマです。

私自身、solaris チームにジョインして、初めて Spanner の Change Streams に触れる機会がありました。Change Streams は、solaris が決済データを遅延なく・欠損なく取り込むために採用している技術です。最初はドキュメントを読んでも「で、実際のコードではどう使うの？」がなかなかイメージできず、既存の実装を読み解きながら少しずつ理解していきました。この記事は、その過程で「最初に知っておきたかったな」と思ったことを、実際の実装例に沿ってまとめたものです。これから Spanner / Change Streams を触る方に、同じ目線で読んでもらえたら嬉しいです。

テーマは、solaris が決済データを遅延なく・欠損なく取り込んでいる仕組みの一端です。具体的には、

Spanner Change Streams をどう読むか（クエリと、返ってくるデータ）
パーティションという考え方
読み取り位置をどう管理しているか
複数 Pod で安全に動かすための排他制御（FOR UPDATE SKIP LOCKED）

あたりを、コードを交えて見ていきます。

全体像

solaris の中では、決済処理そのものを担うサービス（便宜上「決済処理サービス」と呼びます）が決済の Single Source of Truth になっています。このサービスは、カードが使われるたびに、その事実を Spanner にレコードとして書き込んでいきます。

私たちが取り組んでいるのは、「そこに書かれた決済の事実を読み取って、ユーザーに見せる明細に集約する」という部分です。決済処理サービスと、その事実を取り込む側は、別々のマイクロサービスとして動いています。

%%{init: {'theme':'base', 'themeVariables': {'lineColor':'#ffffff','primaryTextColor':'#ffffff','primaryBorderColor':'#ffffff'}}}%%
architecture-beta
    group processor(cloud)["決済処理サービス"]
    service papi(server)["決済取得 API"] in processor
    service spanner(database)[Spanner] in processor

    group ingest(cloud)["solaris 取り込み側"]
    service cdc(server)["変更取り込みサービス"] in ingest
    service queue(cloud)["Pub/Sub"] in ingest
    service tx(server)["明細生成サービス"] in ingest
    service pg(database)["PostgreSQL"] in ingest

    spanner:R --> L:cdc
    cdc:R --> L:queue
    queue:T --> R:tx
    tx:L --> R:papi
    tx:B --> T:pg

ざっくりした流れはこうです。

決済処理サービスが、決済の事実を Spanner に書き込む
solaris の変更取り込みサービスが、Spanner の変更を読み取り、決済を識別する ID を Pub/Sub に流す
明細生成サービスが Pub/Sub を受け取り、その ID をもとに決済処理サービスに問い合わせて実データ（金額など）を取得し、明細に集約する

この記事で深掘りするのは、まさに 2 の「変更取り込みサービス」です。決済には、カードを切った瞬間にリアルタイムに発生する「オーソリ（与信確保）」と、売上が確定して深夜などにまとめて大量に発生する「クリアリング（売上確定）」という性質の異なるデータがあり、その両方を遅延なく・欠損なく拾う必要があります。

なお、こうして「データベースに書き込まれた変更そのものを読み取って取り込む」やり方は、一般には Change Data Capture（CDC）と呼ばれます。本記事では仕組みの中身に集中したいので、この呼び名自体は以降は使いませんが、馴染みのある方は「Spanner Change Streams を使った CDC の実装例」と捉えてもらえれば結構です。

ではこのサービスが、Spanner の変更をどうやって読んでいるのかを見ていきましょう。

Spanner Change Streams をどう読むか

まずは公式ドキュメント

Change Streams そのものの仕様は、公式ドキュメントが一番詳しいです。仕組みをきちんと知りたい方はまずこちらを読むのがおすすめです。ここでは「実際にクエリを投げると、どう書いて、何が返ってくるのか」という、実装する側の目線に絞って紹介します。

クエリの形

Change Streams は、専用の API を subscribe するのではなく、SQL の関数として読みます。READ_<変更ストリーム名>(...) という関数に「いつからいつまでの変更が欲しいか」を渡すと、その期間の変更が行として返ってきます。

SELECT ChangeRecord
FROM READ_PaymentRecordsChangeStream(
  start_timestamp        => @start_timestamp,
  end_timestamp          => @end_timestamp,
  partition_token        => NULL,
  heartbeat_milliseconds => @heartbeat_milliseconds,
  read_options           => NULL
)

補足: ストリーム名・テーブル名は説明のために一般化しています（PaymentRecordsChangeStream は仮名です）。

Go 側では、これをごく普通の Spanner クエリとして実行しているだけです。

stmt := r.buildChangeStreamQuery(streamName, startTimestamp, endTimestamp, partitionToken, heartbeatIntervalMs)
iter := r.client.Single().Query(ctx, stmt)
defer iter.Stop()

ポイントは、「未来に向かってストリームを待ち受ける」のではなく「start 〜 end の時間窓を切り出して読む」モデルだということです。読み終わったら次の窓へ進む、という進め方になります。この「位置を進めていく」という感覚が、あとの「読み取り位置の管理」につながります。

返ってくるデータ

返ってくる ChangeRecord には、大きく3種類が含まれます。

レコード種別	中身
`data_change_record`	実際のデータ変更（INSERT / UPDATE / DELETE）。本当に欲しいのはこれ。
`child_partitions_record`	後述する「パーティション」を辿るための情報。
`heartbeat_record`	「この時刻までは変更がなかった」という情報。

solaris では、このうち data_change_record の中から必要なものだけを取り出して、Pub/Sub に流しています。

パーティションという考え方

上のクエリで partition_token => NULL を渡しているのは、ルートパーティションを読むためです。Change Streams のデータは複数のパーティションに分かれており、ルートを読むと、その結果の中に「子パーティションのトークン」が返ってきます。その子を読むと、さらに孫が返ってくることもあります。

つまりパーティションは木構造になっていて、変更を取りこぼさず読むには、この木を辿りきる必要があります。

flowchart TD
    Root["ルートパーティション<br/>(token = NULL)"]
    A["子パーティション A"]
    B["子パーティション B"]
    A1["孫パーティション A-1"]
    A2["孫パーティション A-2"]

    Root --> A
    Root --> B
    A --> A1
    A --> A2

solaris の実装では、これをキューで辿っています。ルートをキューに入れ、取り出して読み、返ってきた子パーティションをキューの末尾に積む。これを繰り返して木を辿りきります。

// ルートパーティション（空トークン）でキューを初期化
// token が空文字のときは、buildChangeStreamQuery 内で SQL の partition_token => NULL に変換される
queue := []partitionTask{{token: "", startTimestamp: startTimestamp}}

for len(queue) > 0 {
    // キューの先頭を取り出して読む
    task := queue[0]
    queue = queue[1:]

    rawRecords, err := r.fetchRawRecords(ctx, streamName, task.startTimestamp, endTimestamp, task.token)
    if err != nil {
        return nil, err
    }

    filteredRecords, childPartitions, _ := r.extractAndFilter(rawRecords)
    // ... 変更レコードを蓄積 ...

    // 返ってきた子パーティションをキューの末尾に積む
    queue = append(queue, childPartitions...)
}

再帰でも書けそうですが、キューにしておくと走査の状態がスライス1本に収まって見通しがよく、深さも気にしなくて済みます。

読み取り位置をどう管理しているか

Change Streams は「時間窓を切り出して読む」モデルなので、「どこまで読んだか」を自分で覚えておく必要があります。Pod が再起動しても続きから読めるように、solaris ではこの位置を PostgreSQL に保存しています。

テーブルはとてもシンプルで、こんな形です。

erDiagram
    positions {
        varchar stream_name PK "変更ストリーム名"
        timestamp latest_record_timestamp "最後に読んだ位置（タイムスタンプ）"
        timestamp updated_at "更新時刻"
    }

ストリームごとに「最後に読んだタイムスタンプ」を1行持っているだけです。次に読むときは、ここに保存された位置から start_timestamp を決めて、続きを読みにいきます。

ハートビートが効いてくる場面

ここで出てくるのが、先ほど紹介した、返ってくるレコードの中の heartbeat_record です。

これは、決済がまったく発生しない時間帯のために必要なレコードです。

データ変更が0件の時間帯では、「最後に読んだ位置」を変更レコードからは更新できません。すると位置が進まず、次のポーリングでも同じ時間窓を読み直す……ということになりかねません。

そこで solaris の実装では、ハートビートレコードのタイムスタンプも「最新位置」の候補に含めて位置を前進させています。ハートビートは「この時刻までは変更がなかった」という情報なので、変更がなくても「ここまでは確認済み」として位置を進められるわけです。

// ハートビートレコードのタイムスタンプも最新位置の候補にする
func (r *ChangeStreamRepository) extractHeartbeatTimestamp(
    records []*persistence_model.HeartbeatRecord,
    latestTimestamp time.Time,
) time.Time {
    for _, hb := range records {
        if hb.Timestamp.After(latestTimestamp) {
            latestTimestamp = hb.Timestamp
        }
    }
    return latestTimestamp
}

そして、データ変更が0件でも必ず位置を保存するようにしています。

// データ変更がなくても、常に最新タイムスタンプ（ハートビート由来を含む）に位置を更新する。
// これにより、変更がなかった時間窓を毎回読み直すのを防ぐ。
pos.UpdatePosition(readResult.LatestTimestamp)

if err := uc.positionRepo.Save(ctx, pos); err != nil {
    return usecase.PollResult{}, fmt.Errorf("failed to save position: %w", err)
}

複数 Pod で安全に動かす

取り込みサービス自体が止まっても決済そのものが止まるわけではありません。ただ、ここでの取りこぼしや二重処理は決済情報の不整合につながり、それは事故になります。なので「1 Pod で動かして落ちたら終わり」というわけにはいかず、solaris では負荷に応じて最大2 Pod までスケールする構成で動かしています。

すると新しい問題が出てきます。2つの Pod が同じ変更を同時に読んだらどうなるか？ 同じ変更を二重に流してしまいますし、読み取り位置の更新も競合します。冗長化はしたいけれど、処理は片方だけにやってほしい。

これを、専用のリーダー選出の仕組み（Kubernetes の Lease など）を増やさずに、すでに使っている PostgreSQL の行ロックだけで解いています。使っているのは SELECT ... FOR UPDATE SKIP LOCKED です。

読み取り位置の行に対してロックの取得を試み、

ロックを取れた Pod → そのまま処理する
ロックを取れなかった Pod → 待たずに即「0件」が返るので、何もせず一定時間待って再挑戦する

// FOR UPDATE SKIP LOCKED でロック取得を試みる
pos, err := uc.positionRepo.FindByStreamNameWithLock(ctx, uc.streamName)
if err != nil {
    return usecase.PollResult{}, fmt.Errorf("failed to find position: %w", err)
}

// ロックを取得できなかった（別の Pod が処理中）
if pos == nil {
    return usecase.NewLockNotAcquiredResult(), nil
}

return uc.processChangeStreamRecords(ctx, pos)

SKIP LOCKED がポイントです。普通の FOR UPDATE だと、取れなかった Pod はロックが空くまで待たされ、タイムアウトやデッドロックの温床になります。SKIP LOCKED なら「取れないなら即諦めて帰る」ので、待ちが発生しません。

sequenceDiagram
    participant Pod1 as Pod 1
    participant Pod2 as Pod 2
    participant DB as PostgreSQL
    participant Spanner as Change Streams
    participant PubSub as Pub/Sub

    Note over Pod1,Pod2: 両方の Pod が同時に起動

    Pod1->>DB: BEGIN
    Pod1->>DB: SELECT ... FOR UPDATE SKIP LOCKED
    DB-->>Pod1: ロック取得成功・前回位置を返す

    Pod2->>DB: BEGIN
    Pod2->>DB: SELECT ... FOR UPDATE SKIP LOCKED
    DB-->>Pod2: 0件（ロック取得失敗）
    Pod2->>DB: ROLLBACK
    Note over Pod2: 一定時間待機して再挑戦

    Pod1->>Spanner: 変更を読む（前回位置から）
    Spanner-->>Pod1: 変更レコード
    Pod1->>PubSub: publish
    Pod1->>DB: UPDATE position（位置を保存）
    Pod1->>DB: COMMIT（ロック解放）
    Note over Pod1,Pod2: ロックを取れた方が処理を続ける

さらにこの方式は障害にも強いです。ロックを握った Pod が処理の途中で落ちても、トランザクションが切れてロックは自動的に解放されます。すると、もう片方の Pod が次のポーリングでロックを取り、自然に処理を引き継ぎます。専用の仕組みなしで、行ロック1つで擬似的なリーダー選出ができているわけです。

「専用ミドルウェアを足す前に、まず手元の PostgreSQL で素直に解けないか」という発想は、前回の記事金融インフラへと進化するために、変化に強いプロダクトを作る舞台裏で触れている "Just use Postgres!" にも通じるところがあります。solaris の技術選定の考え方について書かれているので、興味がある方は読んでみてください。

取りこぼしに「すぐ気づける」ようにする

ここまで「欠損なく取り込む」ことを目指して設計してきましたが、それでも現実には、Pub/Sub のメッセージが何らかの理由で処理されなかったり、Nack されて滞留したりすることは起こりえます。大事なのは「絶対に起きない」ことよりも、起きたときにすぐ気づけることだと思っています。solaris では2つの層で監視しています。

1. Pub/Sub のメトリクス監視（速い検知）

Pub/Sub の配信状況については、具体的に次のような点を監視しています。

DLQ（デッドレターキュー）にメッセージが溜まっていないか ― 規定回数リトライしても処理できなかったメッセージが入っていれば、どこかで詰まっているサインです
Ack までに時間がかかっているメッセージがないか ― Nack が続いて再配信を繰り返している、処理が遅延している、といった兆候を捉えます

2. 日次の突合

メトリクス監視だけだと、「メッセージ自体は正常に処理されたが、結果としてデータがずれている」ようなケースを取りこぼす可能性があります。そこで solaris では、日次で決済データの突合（reconciliation）バッチを回しています。当月について、

solaris 側の明細を集計した合計額
決済処理サービス側に問い合わせた利用額

を企業単位で突き合わせ、差分がないかを確認します。両者を並列に取得し、一過性のエラーや不一致は数回リトライしたうえで、それでも一致しなければ「確定した不一致」とみなします。

ポイントは、この突合バッチは通知専用で、差分を検知しても solaris のデータを勝手に書き換えないことです。

この2層を組み合わせることで、「パイプラインの異常（速い検知）」と「結果としてのデータ不整合（確実な検知）」の両方をカバーしています。

なぜ Dataflow ではなく自前で読むのか

ここまで読んで、「Spanner Change Streams なら Dataflow を使うのが定石では？」と思った方もいるかもしれません。実際、Google は Change Streams の読み取りに Dataflow（Apache Beam の SpannerIO コネクタ）を使う方式を推奨しており、パーティションの追跡やチェックポイント管理をコネクタ側が引き受けてくれます。

それでも solaris が Pub/Sub ベースの自前実装を選んだのは、Dataflow を入れると専用のワーカーインスタンスの管理やコスト、ジョブの健全性をどう監視するかといった負荷を新たに抱えることになり、取り込みの規模に対して割に合わないと考えたからです。自前実装なら、すでに動いている Go / Pub/Sub / PostgreSQL on Kubernetes の上にそのまま乗り、新しいランタイムを増やさずに済みます。

「複数 Pod で安全に動かす」で触れた "Just use Postgres!" と根は同じで、手元のスタックで素直に解けるなら、まずそれで解くという判断です。

まとめ

Spanner Change Streams を初めて触る目線で、solaris の決済データ取り込みの実装を紹介させていただきました。最後に、この記事で見てきた要点を振り返っておきます。

Change Streams は専用 API ではなく SQL の関数として、時間窓を切り出して読む
データはパーティションに分かれていて、辿りきる必要がある
位置管理はハートビートまで含めて考える（変更がない時間帯にも位置を進める）
FOR UPDATE SKIP LOCKED は、軽量なリーダー選出として使える
「欠損ゼロ」を狙いつつ、「ずれたらすぐ気づける」監視（Pub/Sub メトリクス＋日次突合）まで含めて初めて安心できる

私自身、実装を読み解く中で「最初に知れてよかった」と思ったことばかりでした。

余談ですが、この取り込みの仕組みのおかげで、事情があって本番の DB インスタンスを作り直す必要が生じたときも、リアルタイムに発生し続ける決済データを欠損なく移行しきることができました。位置を保存して続きから読み直せる、という設計の堅牢さに、こういう場面で助けられています。

同じくこれから Spanner / Change Streams を触る方の参考になれば嬉しいです。最後まで読んでいただき、ありがとうございました。

法人カード基盤の外部提供の挑戦技術ブログバトンリレー

各エンジニア組織が法人カード基盤の外部提供の挑戦の中で得た知見を順次公開していきます！技術、開発設計や思想、Tipsなど、幅広くお伝えします。

▼公開予定表（各記事公開後、リンク集になります）

タイトル	チーム	執筆者
UPSIDERのカード基盤を「外部提供できる形」にするまで——マイクロサービス化とマルチブランドアーキテクチャ	法人カード基盤の外部提供全体構想	Mitsui
Spanner Change Streams を自前で読む ― 決済データを欠損なく取り込む	カード管理基盤チーム（Solaris）	Ohsako
1年動き続けるワークフローを壊さずにデプロイする ― Temporal Worker Versioning 導入記	請求基盤チーム（Hermes）	nishiwaki
GCP Error Reportingで「最低限の異常検知」を低コストに実現する	法人審査基盤チーム（Hestia）	Mitomi
複数の銀行データソースを1つに束ねる ― マルチブランド法人カードを支える銀行連携基盤	銀行連携基盤	terry
仮説駆動で取り組む負荷試験計画	processor	Kinsho
AIが実装しレビューする時代に、プロダクトエンジニアは何をするのか	お客様向け管理画面チーム（Iris）	Akari
法人カード基盤の外部提供にも耐えうる不正利用対策基盤の構築	Anti-fraudチーム	Ryutaro
法人カード基盤の外部提供という挑戦決済基盤を「他社にまるごと使ってもらう」QAの話	QAチーム	Naoki