Realtime Updates (HelloInterview) のメモ

この記事で学んだことをメモる。

Patterns

• Real-time Updates のメモ <--- これ 🟢
• Dealing with Contention のメモ
• Multi-step Processes のメモ
• Scaling Reads のメモ
• Scaling Writes のメモ
• Handling Large Blobs のメモ

2 Hops for realtime updates

リアルタイムな更新の伝搬を2ホップに分割して考える。

TL;DR When to use which

2nd Hop / 1st Hop

Practice Problems では、以下を使っていた。

• LeetCode Dashboard は、Simple Polling = Server-Side Polling
• Whats App は、WebSocket x Pub/Sub -> Pub/Sub sharding
• FB Live Comments は、SSE x Pub/Sub -> Pub/Sub sharding -> Mega-Streams(後述)
• Online Auction は、SSE x Pub/Sub
• Google Docs は、WebSocket x Consitent Hashing (to connect all users to the same server)

★ 特に Scalability を求められるのは FB Live Comments.

★ Google doc が、Pub/Sub ではなく Consistent Hashing を使う理由は、Docを編集しているすべてのユーザーを１つのサーバーに繋げ、そのサーバーを　Central Serverとして、変更の Serialization (順序決定) をするため。(ref)

2nd Hop. Server Z --> Client Z

• Short Polling = Frequent updatesの伝搬時に遅延。Frequent updatesの場合、HTTP handshake/terminatino の往復時間がレイテンシーに加算されてしまう。
• Long Polling = 通常のHTTPとインフラ設定は同じ。Frequent updatesの場合、前の更新の復路と新しいコネクションの設立時間がレイテンシーに加算されてしまう。
• ★ Server Sent Event = これも通常のHTTPとインフラ設定は同じ。1回の接続時間は30秒とか。
• Web Socket = bare TCP connection。何時間も繋ぎ続けることがある。
• WebRTC = P2P connections between clients。TODO: 必要ならまとめる。

📝 SSE / WebSocket を使う場合、{ chat_room -> connection[] } のマッピングを、shared memory か local にたてた Redis 内に保存する。１つのSSE / WebSocket connection　は約10KB。
📝 Web Socket コネクションは L4 LoadBalancer を通り続ける (一回繋がったらあとはサーバー間を直接、とはならない。)

📝 負荷の比較

1st Hop. Client A --> Server A --> Server Z

特に Server A --> Server Zの部分。

🟢 Pushing with short polling

これを使うのは、基本的に Hop2 が Short Pollling である場合のみ。

🟢 Use Consitent hashing to keep all connections in one server

ZooKeeperのZnodesの中には、Parameters for Consistent Hashing (Server Membership, VNode Countなど) を保存する。リクエストを正しいサーバーへ導く方法は以下の二通りがある。

• Discovery方式 接続先のバックエンドサーバーの指定を含むアドレスを調べるAPIを作り、クライアントがこれを最初に叩く。その後指定された専用のサブドメイン (e.g. server-001.example.com) をたたく。
• Redirect方式 クライアントは最初にランダムなバックエンドサーバーに接続され、そこから正しいサーバーにリダイレクトされる。

弱点

• シャーディングのサーバーを追加/削除した場合のScaling Transition中にユーザーを別のサーバーへ再接続させるのが大変であること
• TODO: 例えば FB Live Comments で一つの動画に10M usersが入った時の場合など、一つのサーバーにすべてのコネクションが乗り切らなくなる場合どうする？ (ref)

📝 Redirect 方式の図

🟢 ★ Pub/Sub

最も一般的。なぜなら、汎用的、実装が楽、サーバーがどれか落ちても影響が少ないから。

基本的に Redis Pub/Sub で Fire and Forget し、必要に応じて DBからまとめて取得して reconcilliation でいいと思う。

Redis Pub/Sub は 100K messgaes/sec の配信が可能。

TODO: デフォルトチョイスを Redis Pub/Sub over Kafka にする理由をまとめる。
TODO: Kafka を使わなければいけないケースを調べる。

Live Streaming - Mega-Streams パターン

FB Live Comments - Part 2 Metga-Streams で出てきた。ここでまとめる。

例えば World Cup Final の生配信は、500M のビューワーが観戦し、5K comments/sec の勢いでコメントがされることがある。ここまでいくと、全てのメッセージを配信することは必要はなく、およそ 50 comments/sec にサンプリングしたもので良い。そして、ビューワーは個々のメッセージを読むというより、その vibe を楽しむことになる。

🟢 [Good] Pub/Sub x Sampling

送信されたメッセージ数に関わらず、Sampling された一定数のみをクライアントに送れば良い。
表示メッセージのパーソナライゼーションしつつスケーラビリティを考慮すると、以下のように2回に分けてサンプリングすると良い。

• Pub/Sub の前 -> 後続の Redis Pub/Sub と websocket/SSE サーバ の負荷を下げるためにサンプリングする。
• Pub/Sub の後 -> ビューワーごとのパーソナライゼーションを考慮したサンプリングをする。

🟢 [Great] Short Polling x CDN x Sampling

さらにビューワーが増えると、SSE/websocket のコネクションを持ち続けることが負荷になる。
そこでクライアント側による Pull-based モデルに切り替える。サーバーで毎秒50コメントをサンプリングして (TODO: どうやって？) それをCDNに登録する。クライアントは毎秒一回CDNからこれを取得して、これを滑らかに画面上で流れるように表示する。
TODO: より厳密には ring buffer を使うのが適しているらしい。

問題点

• レイテンシー1-2秒かかること。
• コメントの頻度が減ったら、よりリアルタイム性が求められるため、SSE/websocketに切り替える必要がある。TODO: 切り替え方法考える。

c.f. 少数のコメントの全てを大量のビューワーに届ける場合は、Scaling Writes - Batching and Hierarchical Aggregation を使えば良い。

DD "How do you handle connection failures and reconnection?" 🔥🔥

三つ考えることがある。

🟢 サーバーサイドでのコネクション切断検知とゾンビコネクションの削除

WebSocket/SSE のどちらも、クライアント側がコネクションを明示的に終了した場合は問題ない。サーバー側でクライアントとの接続終了を即座に検知できる。

しかし、ネットワーク瞬断やクラッシュによって、コネクションが黙って消えた場合、OSレベルのTCPの仕組みだけでは即座に切断を検知できない。そのため、アプリケーションレイヤーでの制御が必要である。

WebSocket, SSE のどちらもバックグラウンドの定期処理（Heartbeat）で通信を発生させ、異常を検知したらコネクションリソースを削除する方針でOK。

• WebSocket: 応答なし（Timeout）で判定
  • = 定期的に Ping を送り、一定時間内に Pong（応答）がなければゾンビとみなし、connectionを強制的にcloseしてリソースを解放する。
• SSE: 送信不可（Write Error）で判定
  • = 定期的にダミーデータを送信し、相手が消えていた場合に発生する書き込みエラー（Broken Pipe等の例外）をトリガーにしてリソース削除。

🟢 クライアントはどのように正しいサーバーへ再接続されるか

これは初回接続時と同じ。

• Pub/Sub パターン (Stateless Servers) → 任意のサーバーに接続すれば良い。
• Consitent hashing パターン (Stateful Servers) → Discovery方式 か Redirect方式 により接続すべきサーバーを見つける。

🟢 クライアントは接続が途切れていた期間のメッセージをどのように受け取るか

FE側が、BulkFetchMessages API をコールし、最後に受け取ったメッセージ以降を取得する。
クエリパラメータ searchAfter として、message_id (time-sorted UUID like UUIDv7) あるいは厳密にするなら sequence_number を指定する。

📝 WhatsApp では、Inboxテーブルを用意し、{ (device, message) -> is_delivered } のマッピングを管理していた。

DD "What happens when a single user has millions of followers who all need the same update?" 🔥

Scaling Writes - Hierarchical Fanout / Batching and Hierarchical Aggregation パターンを使う。

• Hierarchical Fanout: メッセージやポストの場合は、何も集計する必要がないので、Write/Root Processorは不要。BroadCast Node でPub/Subを多階層にする。
• Batching and Hierarchical Aggregation: Like などの場合は、集計が必要なので、Write Processor / Root Processor のそれぞれで集計し、BroadCast Node でPub/Subを多階層にする。

c.f. 先述した Mega-Stream Pattern は、そもそも作られるメッセージが多すぎてサンプリングが必要な場合のアーキテクチャ。

DD "How do you maintain message ordering across distributed servers?" 🔥🔥🔥

例として、WhatsApp を考える
[edited] WhatsApp は、例として悪かった。Question Breakdow を読んだが、NTP (Network Time Protocol)でベストエフォートで対応する、としか書いていない。

サーバーにおいて発生したイベントをDBに保存する際、created_at カラムにはサーバーの指している現在時刻を使う。しかし、それぞれのサーバー間の時刻にはずれ (Clock Skew) があるため、別サーバで発生したイベントの順序が created_at の順序と異なってしまう場合がある。
TODO: これがどう問題になるのか？

🟢 解決法(1) Vector Clock / Logical Timestamp

Vector Clock や Logical Timestamp を使って複数サーバー間での順序を作成する。
TODO: 理解する

🟢 解決法(2) monotonically increasing sequence number

各イベントに monotonically increasing sequence number を振る。
これを一つのマシンが振っていく。これは以下の方法で可能である。

• メッセージをKafkaの同一パーティションを通し、ここで発行されるOffsetをsequence number として使う。
• サーバーがDB書き込み前に、Redis Counter を叩いて、sequence number を発行する。