PostgreSQL CDC と NATS: リアルタイム アーキテクチャ

ほとんどの CDC Postgres パイプラインは、バイナリ ロジック レプリケーション プロトコルである pgoutputを通過し、それを Kafka に変換する Debezium などのコネクタを通過します。 Aurabase はこのステップを省略します。 aura-realtime サービスは、 、 wal2json プラグインを使用して、 WAL を論理レプリケーション に直接デコードし、中間の変換段階なしで使用可能な JSON を生成します。

-- 存在しない場合は一度作成され、WAL の有効期限が切れた場合は再作成されます
SELECT pg_create_logical_replication_slot(
  'aura_cdc_slot', 'wal2json');

-- ポーリングを繰り返します (バッチが空になったときに 100 ミリ秒バックオフします)
SELECT data FROM pg_logical_slot_get_changes(
  'aura_cdc_slot', NULL, 1000,
  'include-xids', '0', 'format-version', '2',
  'include-schemas', '1', 'include-types', '0');

コードで確認された 3 つの Postgres 前提条件: アプリケーション ロールは REPLICATIONを保持する必要があり、 wal_level = logical がアクティブである必要があり、 max_slot_wal_keep_size が WAL 保持を制限する必要があります。この最後の制限がないと、コンシューマーが遅いため、ディスク ボリュームが無制限に増加します。このサービスはスロットの wal_status も監視します。スロットが lost (この制限を超えて WAL がパージされる) に変化すると、スロットは自動的に再作成されます。その間消費されなかったイベントの損失が想定され、ログに記録されます。

動的テーブル フィルターが 3 秒ごとに更新される、ポーリングごとに 1000 件の変更のバッチ。プロジェクトで明示的にリアルタイムが有効になっているテーブルのみ add-tables を入力します。これにより、サブスクライバーにとって関係のないテーブルの WAL のデコードが回避されます。

Postgres 論理レプリケーション スロットは、一度に 1 つのアクティブ ドライブのみを許容します。同じ aura_cdc_slot の複数のコンシューマーを並行して実行すると、変更が重複するだけでなく、変更の順序が崩れてしまいます。 Aurabase は、WebSocket の水平スケールを犠牲にすることなくこの制約を解決するために、aura-realtime を 2 つの個別のバイナリに分割しました。

# cdc-worker: リーダーに選出され、CDC を立ち上げ、NATS で公開します。 WSサーバーがありません。
[[bin]]
name = "aura-realtime-cdc-worker"

# ws-front: N レプリカ (HPA)、NATS を消費、RLS フィルター、WS/SSE を提供します。
[[bin]]
name = "aura-realtime-ws-front"

cdc-worker は、リース Kubernetes オブジェクト (coordination.k8s.io/v1) を保持することによってのみ、スロットを読み取る権利を取得します。彼はそれを作成しようとするか、有効期限が切れている場合は盗もうとし、寿命の 3 分の 1 が経過すると更新します。リースが失われると (更新が失敗し、所有者が変更されると)、CancellationToken によって進行中の作業が中断され、プロセスは取得ループに戻ります。 Kubernetes の外部では (ローカル テスト中や非クラスター開発中など)、K8s クライアントは接続に失敗し、サービスは「常にリーディング」モードに切り替わります。開発環境では便利ですが、複数のレプリカがある本番環境で忘れた場合は正しくありません。

pub async fn run_with_lease<F, Fut>(cfg: LeaseConfig, work: F) -> anyhow::Result<()>
where F: Fn(CancellationToken) -> Fut, ...
{
  // すべての release_duration_secs / 3 を更新します
  // K8 を除く: ワーク (トークン) は 1 回のみ、トークンはキャンセルされません
}

これは、この種のパイプラインで最も誤解されやすい点です。NATS と JetStream は 2 つの別個のものであり、CDC イベントは永続的な JetStream ストリームを通過しません。 cdc-worker は、Client::publish_with_headers — NATS コア パブ/サブ APIを使用して各イベントをパブリッシュします。これは、JetStream とは別に、永続化やリプレイなしで 最大 1 回 で配信されます。

// cdc-worker がパブリッシュ (ChannelManager::publish_and_relay)
nc.publish_with_headers(subject, headers, payload).await

// ws-front はサブツリー全体 (NatsRelay::start_relay) をリッスンします。
client.subscribe("aura.realtime.>").await

各レプリカ ws-front はサブツリー aura.realtime.>をサブスクライブし、トピックを元のチャネル名にデコードします。次に、接続されている WebSocket のローカル tokio::sync::broadcastにイベントを再発行します。 Aura-Origin ヘッダーには、送信インスタンスの識別子が含まれます。各レプリカは、自身が発行したメッセージを無視します。これにより、中央調整なしで重複が回避されます。

この選択は直接的な結果をもたらします。NATS コアは、長時間持続するキューを持たずに、最大 1 回 (最大 1 回) を配信します。イベントの通過時に ws-front レプリカが NATS から一時的に切断されると、追いつきません。チャネルごとの履歴 (Channel.history、 HISTORY_SIZEによって境界付けられたメモリ内バッファ) は、クラスター レベルではなく、各レプリカでローカルに存在します。 Postgres 論理レプリケーション スロットが上流の永続的な信頼できるソースであり続けるため、これは許容できる妥協策です。使用前に DB の変更が失われないようにするのは、NATS ではなく wal2json です。

JetStream は aura-realtime に存在しますが、用途はまったく異なります。これは、クロスインスタンス プレゼンス KV ストア (aura_presence、メモリ ストア、60 秒 max_age) を提供し、すべての ws-frontレプリカ間で誰がどのチャネルでオンラインであるかを同期します。短期間の共有状態であり、再生する CDC イベントのストリームではありません。

CDC イベントは、チャネルのすべての加入者にそのまま送信されるわけではありません。ポーラーは各変更を CDC_NUM_SHARDS ワーカー (デフォルトでは 4 つ、 project_idによってハッシュ化) の 1 つにルーティングし、最初に pg_policiesをポーリングします。 RLS ポリシーのないテーブルではすべてのサブスクライバー (Supabase の動作) が許可されますが、ポリシーのあるテーブルではサブスクライバーごとのチェックがトリガーされます。

BEGIN;
SET LOCAL ROLE "aura_authenticated";
-- GUC に注入されたサブスクライバの JWT クレーム
SELECT set_config('request.jwt.claims', $1, true);
SELECT EXISTS(SELECT 1 FROM "schema"."table" WHERE id::text = $2);
ROLLBACK;

トランザクションはサブスクライバーの物理的な Postgres の役割を引き受け、その JWT クレームを request.jwt.claims に挿入します。これは、auth.uid() および auth.role() がポリシー側で読み取るものです。次に、このロールの下でラインが表示されたままであることを確認し、書き込みを行わず、実際の RLS 読み取りをキャンセルします。検証された sub_id のみがイベントの allowed_sub_ids に含まれます。空の vec は、フェールクローズでは誰もいないことを意味します。

移行の邪魔になることが多い詳細: wal2json は、テーブルに REPLICA IDENTITY FULLがある場合、 UPDATE または DELETE の完全な内容のみをキャプチャします。これがないと、INSERT のみが復帰します。これが、テーブルでリアルタイムを有効にするエンドポイントがこの ALTER TABLE を自動的に設定する理由です。これは、個別のチェックボックスではなく、リポジトリに対する実際の修正です。

この出願履歴は、理論上の事件ではなく、実際の罠を記録している。 Kubernetes クラスターでデフォルトで使用される Postgres イメージ (pgvector/pgvector ストック コミュニティ イメージ) には、wal2jsonプラグインが含まれていません。これがないと、pg_create_logical_replication_slot('aura_cdc_slot', 'wal2json') は失敗し、クライアント側ではこれを明示的に通知するものはありません。 WebSocket は開いたままで、サブスクリプションは受け入れられますが、DB イベントは発生しません。

RUN apt-get update && \
    apt-get install -y --no-install-recommends \
      postgresql-16-wal2json=2.6-4.pgdg12+1 \
      ...

実際の修正は、このパッケージがインストールされた専用の Postgres イメージを構築して公開し、ストック イメージの代わりに Kubernetes StatefulSet をそのイメージに指定することでした。また、汎用イメージには存在しない、 wal_level=logical、 max_replication_slots 、および max_slot_wal_keep_size を起動引数として明示的に設定します。

cdc-worker が専用の /health エンドポイントで cdc_active フラグ (スロットが正常であることが確認されない限り false) を公開するのは、まさにこのタイプの障害を観察可能にするためです。これは、クライアント側イベントの欠如だけから CDC が停止していると推測するのではなく、注意すべきバイナリ シグナルです。

このメカニズムは単一の POSTGRES_REPLICATION_URL変数を読み取ります。つまり、単一の Postgres サーバーと単一の論理スロットが読み取ります。これは、共有クラスター上のプロジェクトごとのマルチテナント スキーマ モデル (project_<uuid>) と一致しています。このモデルでは、すべてが機能します。単一の cdc-worker は、同じクラスター内のすべてのプロジェクトの変更を確認し、図ごとにルートします。

専用の CNPG クラスター (独自の分離された Postgres インスタンス) 上のプロジェクトの場合、現在使用されているイメージ (CloudNativePG ストック イメージから構築) は wal2jsonではなく pg_graphqlのみを追加します。また、専用クラスターごとに cdc-worker を実行するものはありません。したがって、コードの現在の状態では、ここで説明するリアルタイム CDC は、ロードマップ機能ではなく、実際のアーキテクチャ境界である共有層に焦点を当てたままになります。

このメカニズム全体は SDK からは見えないままです。aurabase-js チェーン可能 API は、イベントが NATS 経由の cdc-worker から来たのか、ローカル開発で分割されていない aura-realtime バイナリから来たのかに応じて変わりません。

const channel = aura.realtime.channel('room-1')
  .on('postgres_changes', {
    event: 'INSERT', schema: 'public', table: 'messages'
  }, (payload) => { /* … */ })
  .subscribe()

ただし、単一のイベントが発生する前に、テーブルを CDC に登録する必要があります。これは作成時に自動的に行われるわけではありません。専用エンドポイント (または Studio の同等のトグル) での service_role 呼び出しは、プロジェクトのプラットフォーム スキーマの realtime_tables にテーブルを書き込み、REPLICA IDENTITY FULL を設定します。

curl -X PUT \
  https://api.aurabase.cloud/v1/db/<project_id>/schema/tables/messages/realtime \
  -H "apikey: <service_role_key>" \
  -H 'Content-Type: application/json' \
  -d '{"enabled": true}'

Supabase と互換性のある SDK サーフェスの残りの部分 (認証、ストレージ、ポリシー RLS) については、 移行ガイドを参照してください。 aura-realtime が単一のワークスペース クレート内で複数のバイナリに分割される方法については、 Cargo アーキテクチャの詳細を参照してください。