Pub/sub

kevyで1つの発行者から多数の購読者へメッセージをファンアウトする方法を説明します。ワイヤ上ではPUBLISH/SUBSCRIBEで、プロセス内では組み込みのStore経由で、そしてkevy-clientのほかの部分と同じURLファサード越しに使えます。

このドキュメントが必要になるとき

1つのライターが0以上のリーダーに*いますぐ*通知したい、そしてリーダーがオフラインの間に届いたメッセージは気にしない。そういう場面がpub/subの出番です：

「全Webワーカーにconfigキャッシュをリフレッシュしろと伝える。」
「あるシャードに書き込まれたばかりの行を、tailしている誰かにストリーミングする。」
「ジョブが着地したらワーカープールを起こす。ジョブ本体はリストに置く。」
「開発ループ：プロデューサスレッドとコンシューマスレッドが同じバイナリ内。Redisサーバーは不要。」

耐久性のあるハンドオフ（リトライ付きジョブキュー、再起動をまたぐファンアウト、メッセージリプレイ）が必要なら、代わりにリストかストリームを使ってください。何がディスクに書かれるかはdocs/persistence.mdを参照してください。

中心となる考え方

pub/subのチャネルとは名前のことです。購読者はその名前（またはグロブパターン）への関心を登録し、同じ名前へのpublishは購読者インデックスを歩いて、マッチした購読者ごとにボディのコピーを1つキューに積みます。ブローカーキューも、オフラインバッファも、ackもありません。publishした瞬間に誰も聴いていなければ、そのメッセージは消えます。

                   publish("news", body)
                          |
                          v
             +-----------------------+
             |  channel "news"       |   <- チャネルごとの購読者インデックス
             |  subscribers: [A,B,C] |
             +-----------------------+
                  |       |       |
                  v       v       v
               sub A   sub B   sub C    <- それぞれが自分のコピーを受ける

内部では、各publishはワイヤフレームを1度だけ構築し、ボディをArcで包み、writevでマッチする全TCP購読者にscatter-gatherします。つまりファンアウトがどれだけ広くても、ボディのバイト列の追加コピーはゼロです。同じチャネル別インデックスが、サーバー接続とプロセス内のSubscriptionハンドルの両方を扱います。

動かしてみる例

`redis-cli`でスモークテスト

動作中のkevyサーバーに対してシェルを2つ開きます：

# シェル 1 — 購読者
$ redis-cli -p 6379 SUBSCRIBE news
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "news"
3) (integer) 1

# シェル 2 — 発行者
$ redis-cli -p 6379 PUBLISH news "hello"
(integer) 1   # 1 人の購読者が受け取った

シェル1に戻ると：

1) "message"
2) "news"
3) "hello"

購読者のいないチャネルへのPUBLISHは(integer) 0を返し、メッセージは捨てられます。これが契約です。「配信を試みた」という類のシグナルは出ません。

URLファサード越しのRust — `kevy-client`

同じ呼び出しの形で、TCPサーバー、名前付きプロセス内バス、永続的なプロセス内ストアのどれでも狙えます。URLを切り替えて再コンパイルするだけで、呼び出し側にmatch scheme { … }は要りません。

use kevy_client::{Connection, Subscriber, PubsubEvent};

fn run(url: &str) -> kevy_client::KevyResult<()> {
    // `news` に対する購読者を開く。バスが最初に返すフレームは subscribe ack なので、
    // ボディをアサートする前にドレインする。
    let mut sub = Subscriber::connect_channels(url, &[b"news"])?;
    let _ack = sub.recv()?;

    let mut conn = Connection::connect(url)?;
    let received = conn.publish(b"news", b"hello")?;
    assert_eq!(received, 1);

    match sub.recv()? {
        PubsubEvent::Message { channel, payload } => {
            assert_eq!(channel, b"news");
            assert_eq!(payload, b"hello");
        }
        other => panic!("unexpected frame: {other:?}"),
    }
    Ok(())
}

// 開発:  名前付きのプロセス内共有バス。
run("mem://app")?;
// 本番: 実際の TCP サーバー。
run("kevy://prod-cache:6379")?;
# Ok::<(), kevy_client::KevyError>(())

スレッドをまたぐ場合もコードは同じです。別々のスレッドから同じURLに対してSubscriberとConnectionを1つずつ開くだけです。mem://<name>のレジストリが両端に同じバッキングバスを渡すので、プロデューサスレッドがConnection::publishし、コンシューマスレッドがsub.recv()でブロックします。

`kevy-embedded`経由のプロセス内利用

組み込み側のコードがすでにStoreを持っているなら、URLの間接層を飛ばして直接バスと話せます：

use kevy_embedded::{Config, PubsubFrame, Store};

let store = Store::open(Config::default().with_ttl_reaper_manual())?;

// 購読者は受信キューを所有する。
let sub = store.subscribe(&[b"jobs"]);
let _ack = sub.recv()?; // PubsubFrame::Subscribe

// `store` のどのクローンも同じバスに届く。
let writer = store.clone();
assert_eq!(writer.publish(b"jobs", b"compute-pi"), 1);

match sub.recv()? {
    PubsubFrame::Message { channel, payload } => {
        assert_eq!(channel, b"jobs");
        assert_eq!(payload, b"compute-pi");
    }
    other => panic!("unexpected frame: {other:?}"),
}
# Ok::<(), kevy_embedded::KevyError>(())

Store::cloneは安価（Arcのカウントを増やすだけ）なので、典型的な形は「各スレッドにstore.clone()を渡し、必要になったときにpublishやsubscribeをさせる」というものです。購読者のdropはアトミックに登録解除されるため、コンシューマスレッドがパニックしてもインデックスにゾンビエントリは残りません。

パターン購読

PSUBSCRIBEはグロブを登録し、それにマッチするすべてのチャネルのメッセージを受け取ります。グロブ構文（*、?、[abc]）は、KEYSとSCANが使うマッチャと同じです。

use kevy_client::{Connection, Subscriber, PubsubEvent};

let mut sub = Subscriber::connect("mem://signals")?;
sub.psubscribe(&[b"news.*"])?;
let _ack = sub.recv()?;            // PubsubEvent::Psubscribe

let mut conn = Connection::connect("mem://signals")?;
conn.publish(b"news.tech", b"breaking")?; // マッチ
conn.publish(b"weather",   b"sunny")?;    // マッチしない

match sub.recv()? {
    PubsubEvent::Pmessage { pattern, channel, payload } => {
        assert_eq!(pattern, b"news.*");
        assert_eq!(channel, b"news.tech");
        assert_eq!(payload, b"breaking");
    }
    other => panic!("unexpected frame: {other:?}"),
}
# Ok::<(), kevy_client::KevyError>(())

チャネル購読と、それにマッチするパターン購読の両方を持つ購読者は、コピーを2つ受け取ります（Messageが1つとPmessageが1つ）。publishごとの重複排除が抑止するのは「同じSubscriptionが同じチャネルインデックスに2回並んでいる」タイプの重複だけで、チャネルとパターンの重なりは抑止しません。

Keyspace notifications（キー空間通知）

kevyはキー空間の変更をRedis流にpub/subで告知できます。user:42への書き込みは__keyspace@0__:user:42（ペイロード＝イベント名、例：set）と__keyevent@0__:set（ペイロード＝キー）の片方または両方を発火します。kevyはDB 0だけを提供するので、チャネル名は常に@0を持ちます。

デフォルトはオフです——フラグ文字列が空なら、書き込みが払うのはアトミックロード1回とスキップだけです。設定ファイルで有効化します。

[notification]
notify_keyspace_events = "KEA"   # everything, both channels

フラグ文字列はRedisの慣例に従います——チャネルが先、イベントクラスが後です。

フラグ	意味
`K`	`__keyspace@0__:<key>`チャネルへpublish
`E`	`__keyevent@0__:<event>`チャネルへpublish
`g`	汎用コマンド——`DEL`、`EXPIRE`、`PERSIST`、`RENAME`など
`$`	文字列コマンド——`SET`、`INCR`、`APPEND`、`MSET`など
`l`	リストコマンド
`s`	セットコマンド
`h`	ハッシュコマンド
`z`	ソート済みセットコマンド
`t`	ストリームコマンド——`XADD`、`XTRIM`、`XGROUP`など
`x`	`expired`イベント——TTL付きキーの除去（アクセス時の遅延削除、またはリーパーによる）
`e`	`evicted`イベント——`maxmemory`圧による除去
`n`	`new`イベント——キー空間へのキー追加
`A`	`g$lshztxe`のエイリアス（Redisに合わせ、`n`以外の全クラス）

何かを発火させるには、K／Eの少なくとも一方＋少なくとも1つのイベントクラスが必要です。文字列中の未知の文字は起動時のconfigエラーです——タイポしたフラグ文字列は、イベントを黙って落とす代わりに入場を拒否されます。購読は普通のパターン購読で行います。

redis-cli -p 6379 PSUBSCRIBE '__keyevent@0__:*'

通知の配送は、このページの他のすべてと同じat-most-onceのpub/subバスに乗ります。publish時点で購読者がいなければ、イベントは消えます。変更を取りこぼしてはいけないコンシューマはCDCフィード（docs/cdc.md）に乗せてください。あちらはカーソル付きでリプレイ可能です——キー空間通知は「起きろ」の合図であって、台帳ではありません。

URLバックエンド表

URL	バッキングストア	openをまたいで共有される？	プロセスをまたいで見える？
`mem://`	プロセス内、匿名	いいえ — openのたびに新しい`Store`	いいえ
`mem://<name>`	プロセス内、名前付きレジストリ	はい — 同じ`<name>` ⇒ 同じ`Store`	いいえ
`file:///abs/path`	プロセス内 + AOF/スナップショット	はい — 同じパス ⇒ 同じ`Store`、永続	いいえ
`kevy://host[:port][/db]`	TCPのkevyサーバー	openごとに1ソケット、サーバー側でファンアウト	はい
`redis://host[:port][/db]`	TCP — `kevy://`のエイリアス	同上	はい
`tcp://host[:port]`	TCP — 生。先頭の`SELECT`なし	同上	はい

匿名のmem://は発行されたメッセージを受け取れません。同じバッキングStoreにほかの誰も到達できないため、Subscriber::connect_channelsはKevyError::Unsupportedで拒否します。publishするつもりがあるなら、常にmem://<some-name>を使ってください。

rediss://、kevys://、redis://user:pass@…も同じ理由で拒否されます。kevyはTLSもAUTHもなしで出荷されるからです。どちらかが必要なら、ネットワーク境界でstunnelとIP許可リストを前段に置いてください。

mem://<name>とfile:///のレジストリはプロセス単位です。無関係な2つのOSプロセスが同じ名前を開いても、見えるのは独立した2つのバスです。プロセスをまたいだ配信が欲しいなら、kevyサーバーを立てて両側からkevy://host:portを開いてください。

トレードオフと限界

at-most-once配信。 フレームの途中で切断した購読者は、そのフレームを失います。購読者ごとの耐久カーソルも再配信もありません。フレームが重要なら、リストかストリームに永続化し、pub/subは「起こす」シグナルとしてだけ使ってください。
オフラインバックログなし。 購読者ゼロのpublishは0を返してボディを破棄します。切断中に見逃した分を購読者に追いつかせるバッファはありません。
購読者のバックプレッシャは購読者単位で、グローバルではありません。 各購読者は自分専用の有界キューを持ちます。遅いコンシューマは自分のキューを埋め、その後はフレームを落とすか、TCPならサーバーのクライアント出力バッファポリシーによって切断されます。publishパスは送信前にバスのミューテックスを手放すので、遅いリスナー1人が無関係なチャネルのpublishを止めることはできません。その代わり、発行者へバックプレッシャを掛けることもできません。
Linuxのwritev上限。 Linuxでは、writevが1回の呼び出しでカーネルに渡せるiovecエントリは最大IOV_MAX = 1024です。サーバーは購読者ごとのフレームヘッダと共有ボディのArcをiovecにまとめます。1購読者あたりiovecを3つ使うため、チャネルあたり約340購読者を超えるファンアウトでは、サーバーが自動的に複数回のwritev呼び出しに分割します。この上限はソフトな性能の天井として現れるだけで、配信失敗にはなりません。
購読中のクライアントは制限されます。 Subscriberコネクションはpub/sub以外のコマンドを拒否します。kevy-clientが発行者と購読者を、同じURLを共有する別々の2つの型として公開しているのはこのためです。

運用イントロスペクション

標準のPUBSUB管理サブコマンドはTCPサーバーでもURLファサードでも動きます。呼び出すときはSubscriberではなく通常のConnectionを開いてください。

サブコマンド	戻り値
`PUBSUB CHANNELS [pat]`	購読者が1人以上いるチャネルの配列。オプションでグロブフィルタ可。
`PUBSUB NUMSUB [ch …]`	指定した各チャネルについて`channel, count`のペアを交互に返す（存在しなければ0）。
`PUBSUB NUMPAT`	整数。全クライアントを通じて登録されている`PSUBSCRIBE`パターンの異なり数。

$ redis-cli -p 6379 PUBSUB CHANNELS '*'
1) "news"
2) "jobs"
$ redis-cli -p 6379 PUBSUB NUMSUB news jobs missing
1) "news"
2) (integer) 3
3) "jobs"
4) (integer) 1
5) "missing"
6) (integer) 0
$ redis-cli -p 6379 PUBSUB NUMPAT
(integer) 2

3つともシャードごとのpub/subレジストリに対するO(channels)またはO(args)のポイントルックアップなので、監視エージェントからポーリングしても安全です。

FAQ

publishの後で接続した購読者にメッセージは届きますか？ いいえ。pub/subにリプレイはありません。購読者インデックスはpublish時点で参照されます。後から購読した者に見えるのは、自分のsubscribe ackが着地した*後*に発行されたフレームだけです。

PUBLISHは購読者がドレインするまで発行者をブロックしますか？ いいえ。発行者のpublish呼び出しは、ボディがマッチする全購読者の購読者別キューに積まれた時点（TCP購読者の場合はさらに各ソケットの書き込みキューにスケジュールされた時点）で戻ります。遅い購読者が詰まらせるのは自分のキューであって、あなたのキューではありません。

1つのSubscriberをasyncタスク間で共有できますか？ はい。Arcで包んでrecv呼び出しをspawn_blockingしてください。受信ミューテックスがブロッキング待機を直列化するので、各フレームはちょうど1つのタスクに配信されます。本当のブロードキャストファンアウト（全タスクが全フレームを見る）が欲しければ、タスクごとにSubscriberを1つ開いてください。開くのは安価です。完全なasyncパターンはdocs/async.mdを参照してください。

なぜテストではメッセージより先にsubscribe ackが見えるのですか？ バスは順序付きですが、各SUBSCRIBE/PSUBSCRIBEは、そのチャネルの最初のボディフレームより*先に*ackフレームをキューに積みます。ペイロードをアサートする前に、sub.recv()?を1回呼んでackをドレインしてください。これはredis-cliのワイヤ上の挙動とも一致します。

pub/subにクラスタルーティングは必要ですか？ いいえ。pub/subのファンアウトはプロセスレベルで、スロットルーティングされません。どのシャードのポートでpublishしても、同じプロセス内の全シャードのポートの全購読者に届きます。任意のシャードポートへのConnection::connect("kevy://host:port")で十分です。*キー空間*コマンドが使うスロットルーティングについてはdocs/cluster.mdを参照してください。