PostgreSQLの並行配送・SQL・履歴保持を改善

.github/workflows/test.yml

@@ -2,18 +2,29 @@ name: Plugin
 on: [push, pull_request]
 permissions:
   contents: read
+concurrency:
+  group: plugin-${{ github.event_name }}-${{ github.ref }}
+  cancel-in-progress: ${{ !startsWith(github.ref, 'refs/tags/') }}
 jobs:
   test:
     runs-on: ubuntu-24.04
+    strategy:
+      fail-fast: false
+      matrix:
+        postgres: ['16', '17', '18']
+    env:
+      POSTGRES_IMAGE: postgres:${{ matrix.postgres }}-alpine
     steps:
       - uses: actions/checkout@v4
+      - run: python3 -m unittest discover -s tests/release -v
       - uses: dtolnay/rust-toolchain@stable
         with:
           components: rustfmt, clippy
       - run: cargo fmt --all --check
       - run: cargo clippy --all-targets --locked -- -D warnings
       - run: cargo test --locked
       - run: bash scripts/test-postgres.sh
+      - run: python3 scripts/bench-postgres.py --plans-only --image "$POSTGRES_IMAGE" --output artifacts/query-plans.json
       - run: cargo build --release --locked
       - uses: actions/checkout@v4
         with:
@@ -22,7 +33,29 @@ jobs:
           path: .packaging
           persist-credentials: false
       - run: python3 .packaging/scripts/package-plugin.py target/release/amitoki-plugin-postgres dist
+      - run: cp schema.sql dist/schema.sql
       - uses: actions/upload-artifact@v4
         with:
-          name: amitoki-plugin-postgres-x86_64-linux
+          name: amitoki-plugin-postgres-x86_64-linux-pg${{ matrix.postgres }}
           path: dist/
+  publish:
+    if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags/v')
+    needs: test
+    runs-on: ubuntu-24.04
+    timeout-minutes: 10
+    permissions:
+      contents: write
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+      - uses: actions/download-artifact@v4
+        with:
+          name: amitoki-plugin-postgres-x86_64-linux-pg17
+          path: dist
+      - name: Verify the package and publish the version tag
+        env:
+          GH_TOKEN: ${{ github.token }}
+          RELEASE_TAG: ${{ github.ref_name }}
+          RELEASE_REPOSITORY: ${{ github.repository }}
+        run: python3 scripts/publish-release.py --directory dist --tag "$RELEASE_TAG" --repository "$RELEASE_REPOSITORY"

.gitignore

@@ -1,5 +1,7 @@
 /target/
 /dist/
+/artifacts/
+__pycache__/
 .env*
 *.der
 *.pem

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "amitoki-relay-postgres"
-version = "0.1.2"
+version = "0.2.0"
 edition = "2021"
 
 [dependencies]

README.md

@@ -4,12 +4,12 @@ PostgreSQLを使うamitokiの外部プロセス型プラグイン。本体の再
 
 ## 利用する
 
-amitoki 0.3以降を使う。公開リポジトリなので、公式配布物の取得にGitHubトークンは不要。
+amitoki 0.2以降を使う。公開リポジトリなので、公式配布物の取得にGitHubトークンは不要。
 
 ```bash
-amitoki plugin add postgres
-amitoki plugin configure postgres --set connection_env=AMITOKI_POSTGRES_URL --set max_connections=4
-amitoki plugin validate postgres
+amitoki plugin relay add https://github.com/amitoki/amitoki-plugin-postgres
+amitoki plugin relay configure postgres --set connection_env=AMITOKI_POSTGRES_URL --set max_connections=4
+amitoki plugin relay validate postgres
 read -r -s -p 'PostgreSQL接続文字列: ' AMITOKI_POSTGRES_URL
 printf '\n'
 export AMITOKI_POSTGRES_URL
@@ -35,13 +35,19 @@ max_connections = 4
 replay_window_ms = 4000
 ```
 
-`replay_window_ms`は初回登録前のフレームを何ミリ秒分受け取るか。再起動時は値に関係なく既存の未ACKキューを引き継ぐ。
+`replay_window_ms`は初回登録前のフレームを何ミリ秒分受け取るか。基準は登録ロック取得後の再生SQL開始時刻。0では履歴を取り込まない。再起動時は値に関係なく既存の未ACKキューを引き継ぐ。
 
 ## 配送の扱い
 
-フレームと各ノードの未処理キューを同じトランザクションで保存する。receiveは非破壊で、NICへの注入後にACKする。同じUUIDの再送とACKは冪等。登録・送信の競合はchannel単位のadvisory lockで防ぎ、同一node_idの多重起動は専用DB接続で拒否する。
+フレームと各ノードの未処理キューを同じトランザクションで保存する。receiveは非破壊で、NICへの注入後にACKする。同じUUIDの再送とACKは冪等。送信同士はchannelの共有advisory lockで並行実行し、ノード登録とは排他にする。DBの既定に関係なく登録・送信はREAD COMMITTEDで実行する。同一node_idの多重起動は専用DB接続で拒否する。
 
-NIC注入とACKの間のクラッシュは重複し得る。DBの保存済みデータは再起動後も残るが、収集直後の本体メモリは永続化しない。DBの保存期限・掃除は自動化していない。
+単一バッチ内の入力順を維持する。別の送信が先にcommitすることはあり、全送信元を通じたcommit順は保証しない。重なるUUIDを逆順で送った場合などのDBデッドロックは、再試行可能エラーで返す。本体は同じUUIDのバッチを再送できる。
+
+NIC注入とACKの間のクラッシュは重複し得る。DBの保存済みデータは再起動後も残るが、収集直後の本体メモリは永続化しない。保持期限を設定すると、削除済みUUIDの再送は新規フレームになる。
+
+## 保持期限とDB更新
+
+自動削除は既定で無効。利用する場合は、channel内の全ノードで`retention_ms`を指定する。設定例とDB更新手順は[保持期限と運用](docs/retention.md)を参照。既存DBでは、更新した実行ファイルから取り出した`schema.sql`をプラグインの起動前に再適用する。ノード設定用の列と未ACK確認用の索引を追加し、既存フレーム・未ACKは維持する。
 
 ## 開発・試験
 
@@ -60,8 +66,24 @@ cargo build --release --locked
 
 Dockerを実行できるユーザで試験する。DB試験は一時コンテナを終了時に削除する。公開SDKのGit revisionとCargo.lockを固定している。
 
+送信・受信・ACKの性能とSQL実行計画は[コンテナでの検証手順](docs/benchmark.md)で測定する。検証用の一時DBで、ノード数・パケット長・バッチサイズを変えて全フレームを照合する。
+
 配布物は本体の`scripts/package-plugin.py target/release/amitoki-plugin-postgres dist`で生成する。CIも同じスクリプトを使用する。Linux x86_64向けの初回配布はUbuntu 24.04でビルド・試験した。
 
+## リリースする
+
+`Cargo.toml`と`Cargo.lock`の版を更新し、`docs/releases/v<版>.md`へ更新手順を追加する。mainへマージしてCIが成功したcommitに注釈タグ`v<版>`を付けてpushする。タグのCIがPostgreSQL 16・17・18の試験を完了すると、実行ファイル・manifest・SQLの一致を検証し、SHA256SUMSとともにGitHub Releasesへ公開する。正式公開前にアップロード済み全ファイルのSHA256を照合する。公開済み版を上書きしない。
+
+```bash
+git switch main
+git pull --ff-only
+# 作業ツリーがクリーンで、対象commitのCIが成功していることを確認する。
+git tag -a v0.2.0 -m 'PostgreSQLプラグイン0.2.0を公開'
+git push origin v0.2.0
+```
+
+公開前のdraftで止まった場合は、原因を確認して同じタグのCIを再実行できる。コード修正が必要な場合は新しい版を作り、既存タグは変更しない。
+
 ## stegrdb版から移行する
 
 旧タグとリリースは保持している。amitoki版は実行ファイル・crate・環境変数の接頭辞をamitokiへ変更した。本体の[移行手順](https://github.com/amitoki/amitoki/blob/main/docs/amitoki-migration.md)に従って設定とプラグインを配置する。

docs/benchmark.md

@@ -0,0 +1,66 @@
+# コンテナで配送性能を検証する
+
+検証用のPostgreSQLだけを作り、全フレームの内容とACK完了を照合する。既存DBの接続情報は使わない。プラグインのRust実装とDBの性能を測るため、NIC・本体のパイプライン・外部プラグインのMessagePack通信は測定に含まない。
+
+## 実行
+
+Rust・Python 3・Dockerを準備する。Rust/Dockerの導入は[README](../README.md#開発試験)を参照。
+
+```bash
+cargo build --release --example relay_bench --locked
+python3 scripts/bench-postgres.py \
+  --output artifacts/postgres-bench/result.json
+```
+
+各条件を3回測る。コンテナは4CPU・2GiB、ホストのloopbackのランダムポートを使う。既定イメージは`postgres:17-alpine`で、結果には実際のイメージIDとPostgreSQL版を記録する。比較を再現する場合は`--image postgres@sha256:...`で同じイメージを指定する。
+
+`fsync`と`synchronous_commit`を無効化しない。コンテナの通常ストレージを使い、tmpfsへDBを置かない。ホストのディスクや同時負荷によって速度は変わる。例外・中断時も、このコマンドが作ったコンテナと匿名volumeを削除する。
+
+## 測定条件と結果
+
+- 2・3ノードすべてが同時に送信し、他ノードの全フレームを受信・照合・ACKする。
+- フレーム長64・1400バイト、送信バッチ1・32・128件。受信は最大128件。
+- 本測定前に各ノード1024件を送信し、統計を更新してから受信する。ウォームアップは速度に含めない。
+- `--warmup-frames 0`では空テーブルの統計から始める。投入直後の実行計画の影響を含むため、通常の測定値と混ぜない。
+- 速度は「元の送信フレーム数 ÷ 全ノードのACK完了時間」。複数宛先への配送数は`verified_deliveries`に別記する。
+- バッチ送信のp50・p95・p99、ACK後にDBに残るフレーム数、未ACK数を保存する。保存フレーム数にはウォームアップを含む。
+
+各条件の後、検証用データを削除する。この削除時間は配送速度に含めない。通常の比較では保持期限による自動削除を無効にする。
+
+## 改善候補との比較
+
+比較用ビルドを別の出力先で作り、両方の実行ファイルを指定する。
+
+```bash
+python3 scripts/bench-postgres.py \
+  --binary /absolute/path/current/relay_bench \
+  --compare-binary /absolute/path/candidate/relay_bench \
+  --output artifacts/postgres-bench/comparison.json
+```
+
+比較ビルドでは`CARGO_TARGET_DIR`を分ける。同じpackage名・版のworktree同士で出力先を共有すると、成果物を取り違える場合がある。スクリプトは実行ファイルのSHA256を記録し、同一ファイル同士の比較を拒否する。実行順は繰り返しごとに交替する。
+
+## SQL実行計画だけを調べる
+
+```bash
+python3 scripts/bench-postgres.py --plans-only \
+  --baseline-queries /absolute/path/before/src/queries.rs \
+  --output artifacts/postgres-bench/query-plans.json
+```
+
+受信SQLは1万件のフレームと128件の未ACKで、統計更新前後の計画を比較する。初期条件では、以前のchannelの統計を残したままデータを入れ替え、新channelの件数が少なく見積もられる状況を再現する。時間の閾値ではなく、履歴を繰り返し全走査していないことを実行計画の実走査件数で検証し、PostgreSQL 16/17/18のCIでも実行する。初回再生SQLは25万件の古い履歴で測定する。`--baseline-queries`には比較元の`src/queries.rs`を指定し、省略時は現行SQLだけを測る。比較元のファイルは`git show <revision>:src/queries.rs > before-queries.rs`で取得できる。配送するUUID・本文・順序の一致も確認する。
+
+`EXPLAIN ANALYZE`の結果には実行時間・走査件数・バッファ参照を保存する。実際の稼働中はパラメータ、prepared statementの計画、データ分布、統計の鮮度で計画が変わるため、この固定データでの結果だけで常時高速化を保証しない。
+
+初回再生の基準時刻にはSQL開始時刻で固定される`statement_timestamp()`を使う。[PostgreSQLの時刻関数](https://www.postgresql.org/docs/17/functions-datetime.html#FUNCTIONS-DATETIME-CURRENT)の違いに従い、インデックスで期間を絞れるようにしている。
+
+## 接続を維持した耐久試験
+
+```bash
+python3 scripts/bench-postgres.py --soak-seconds 300 --retention-ms 5000 \
+  --output artifacts/postgres-bench/soak.json
+```
+
+同じ測定プロセス内で3つのRelayとDB接続を維持し、1400バイト・バッチ128件の送受信を繰り返す。すべての内容とACKを照合し、約10秒ごとのRSS、残存フレーム数、未ACK数、表と索引の使用量を記録する。上の5秒保持は掃除を短時間で試験するための値で、本番の推奨保持期間ではない。`--retention-ms 0`では履歴を残したまま測る。耐久試験では掃除間隔を100ミリ秒にする。
+
+各ラウンドの速度と、生成・観測を含む全体時間を別に記録する。10秒のサンプルはラウンド完了後なので、負荷が高いと記録間隔は延びる。RSSには生成・照合用メモリを含み、DBの常駐メモリを含まない。5分の成功だけで日単位のメモリ安定性やDB容量上限は保証しない。

docs/releases/v0.2.0.md

@@ -0,0 +1,27 @@
+# PostgreSQLプラグイン 0.2.0
+
+同じchannelへの送信を並行処理し、履歴が増えた場合や統計が古い場合の受信・再生SQLを改善しました。ACK済みフレームの保持期限も設定できます。
+
+- ノード登録との排他を維持したまま送信を共有ロックへ変更し、登録・送信の分離レベルをREAD COMMITTEDに固定しました。
+- 受信は配送対象の主キーを参照し、古い統計による履歴の全走査を避けます。初回再生には期間索引を使います。
+- `retention_ms`・`cleanup_interval_ms`・`cleanup_batch_size`を追加しました。自動削除は既定で無効です。全登録ノードが有効にした場合だけ、最長の保持・再生期間を過ぎたACK済みフレームを削除します。未ACKは保持します。
+- 古いACKが、期限切れ削除後に同じUUIDで登録された新しい配送を消さないようにしました。
+- PostgreSQL 16・17・18の実DB試験と実行計画の回帰検証をCIで実行します。タグのCI成功後、実行ファイル・manifest・SQL・SHA256SUMSを公開します。
+
+3ノード・1400バイト・128件バッチの比較では、合計約54.7Mbpsから179.1Mbpsへ改善しました。PostgreSQL 17.11、4CPU/2GiBのコンテナ、loopback・TLSなしで各条件3回の中央値です。本体・IPC・NICは含みません。72試行で884,736配送、保持を有効にした約5分の試験で3,366,912配送の内容とACKを照合しました。
+
+## v0.1.2から更新する
+
+本体はamitoki 0.2以降を使えます。本体の再ビルドは不要です。対象DBを使うamitokiを停止してから、プラグインを更新します。
+
+```bash
+amitoki plugin relay update postgres
+~/.local/share/amitoki/plugins/postgres/amitoki-plugin-postgres --schema > schema.sql
+psql "$AMITOKI_POSTGRES_URL" --set ON_ERROR_STOP=1 -f schema.sql
+```
+
+`AMITOKI_PLUGIN_DIR`・`XDG_DATA_HOME`を使う場合は、その配置先からSQLを取り出してください。接続変数名を変更している場合も読み替えてください。スキーマ更新後にamitokiを起動します。既存データと未ACKは維持します。大きな既存DBでの索引作成と、以前の版へ戻す手順は[保持期限と運用](https://github.com/amitoki/amitoki-plugin-postgres/blob/v0.2.0/docs/retention.md)に記載しています。
+
+保持期間は容量上限ではありません。停止したノードの未ACKは残ります。DB再起動で占有ロックを失った場合には本体の明示再起動が必要です。
+
+配布対象はLinux x86_64です。P2Pプラグインと中継IPC v1は変更していません。