なぜWebSocketsで長い投票を選択したのか

HTTP Long Pollingを使用して、node.js、TypeScript、およびPostgreSQLを使用してリアルタイムアップデートを実装する方法を学びます。 WebSocketのないスケーラブルなリアルタイムシステムを構築するための実用的なガイド。

Nadeesha Cabral の上 04-01-2025

Node.jsとTypeScriptを使用してリアルタイムシステムを構築する多くのチームと同様に、リアルタイムの更新を大規模に処理する方法を模索しています。当社のシステムは、新しいジョブ（エージェントが発行したツールコール）のポストグレスクル支援コントロールプレーンを絶えずポーリングしている数百のワーカーノードを処理し、エージェント自体が実行およびチャット状態の更新のために継続的にプルされます。 Websocketsの探索として始まったものは、驚くほど効果的な「古い学校」ソリューション：HTTP Postgresによる長いポーリングに導かれました。

課題：規模のリアルタイムアップデート

node.js/typescriptバックエンドは、2つの主な課題に直面しました。

1. ワーカーノードの更新：node.js / golang / c＃sdksを実行している数百のワーカーノードが利用可能になり次第、新しいジョブについて知る必要があり、Postgresデータベースを削除しないクエリ戦略が必要です
2. エージェント状態の同期：エージェントには、実行とチャット状態に関するリアルタイムの更新が必要でした。これを効率的にストリーミングする必要がありました。

長いポーリングとウェブソケット：リフレッシャー

投票期間はどれくらいですか

sequenceDiagram
    participant Client
    participant Server
    participant Database
    
    Client->>Server: Request new data
    
    alt Data available immediately
        Server->>Database: Check for data
        Database-->>Server: Return data
        Server-->>Client: Return response immediately
    else No data available
        Server->>Database: Check for data
        Database-->>Server: No data
        Note over Server: Hold connection
        loop Check periodically
            Server->>Database: Poll for new data
            Database-->>Server: New data arrives
        end
        Server-->>Client: Return response
    end
    Client->>Server: Next request begins

アプローチ間の重要な違いは、単純な列車の類推で理解できます。

短い投票は、スケジュールに従って厳密に出発する列車のようなものです。乗客がいるかどうかに関係なく、固定間隔でステーションを離れます。一方、WebSocketsは、常に乗客を輸送するための専用の列車ラインを用意しているようなものです。

長い投票？それは、出発する前に少なくとも1つの乗客ボードまで駅で待つ列車のようなものです。特定の時間（TTL）以内に乗客が現れない場合、その後、空のままになります。このアプローチは、両方の世界で最高の世界を提供します – データ（乗客）がある場合はすぐに出発し、そうでない場合は効率的なリソースの使用です。

技術的には：

1. 短い投票で、サーバーはデータがあるかどうかにかかわらずすぐに応答します
2. 長い投票で、サーバーは次のいずれかまで接続を開いています。
   • 新しいデータが利用可能になります
   • タイムアウトに到達する（TTL）

私たちの実装ディープダイビング

node.jsの実装を分解しましょう。

export const getJobStatusSync = async ({
  jobId,
  owner,
  ttl = 60_000,
}: {
  jobId: string;
  owner: { clusterId: string };
  ttl?: number;
}) => {
  let jobResult: {
    service: string;
    status: "pending" | "running" | "success" | "failure" | "stalled";
    result: string | null;
    resultType: ResultType | null;
  } | undefined;

  const start = Date.now();

関数は受け入れます：

• jobId：私たちが追跡している仕事のための一意の識別子
• owner.clusterId：マルチテナンシーのクラスター識別子
• ttl：ミリ秒単位での時間（デフォルトは60秒）

ポーリングループ

  do {
    const [job] = await data.db
      .select({
        service: data.jobs.service,
        status: data.jobs.status,
        result: data.jobs.result,
        resultType: data.jobs.result_type,
      })
      .from(data.jobs)
      .where(and(eq(data.jobs.id, jobId), eq(data.jobs.cluster_id, owner.clusterId)));

    if (!job) {
      throw new NotFoundError(`Job ${jobId} not found`);
    }

    if (job.status === "success" || job.status === "failure") {
      jobResult = job;
    } else {
      await new Promise(resolve => setTimeout(resolve, 500));
    }
  } while (!jobResult && Date.now() - start 

重要な側面：

1. ループは次のように継続します。
   • 最終的なステータスが得られます（success または failure））
   • TTLタイムアウトにヒットします
2. チェック間で500ms遅延を使用して、データベースのハンマーを防ぐ
3. データベースクエリは、適切なインデックスをオンにして最適化されています id そして cluster_id

エラー処理と応答

  if (jobResult) {
    return jobResult;
  } else {
    throw new JobPollTimeoutError(`Call did not resolve within ${ttl}ms`);
  }

関数は次のように終了します：

1. 結果が見つからなかった場合、タイムアウトエラーをスローします
2. 成功した場合、ジョブの結果を返す

データベースの最適化

このパターンが効率的に機能するには、適切なポストグレスインデックス作成を実装する必要があります。

CREATE INDEX idx_jobs_status ON jobs(id, cluster_id);
CREATE INDEX idx_jobs_lookup ON jobs(status) WHERE status IN ('success', 'failure');

これにより、頻繁なポーリングクエリが高速であり、データベースに不必要な負荷をかけないようにします。

長い世論調査の隠された利点

長い世論調査の最も説得力のある側面の1つは、構築する必要がないことです。これが私たちが避けたことです：

観察可能性は変更されていません

最大の勝利の1つは、WebSocketsの観測可能性スタックを変更する必要がないことです。すべての標準のHTTPメトリックは、すぐに機能し、既存のロギングパターンは必要なことを正確に行います。永続的な接続を監視したり、WebSocket状態に追加のロギングを実装する新しい方法を見つけ出す必要はありません。

認証のシンプルさ

私たちは、着信WebSocket接続のための新しい認証メカニズムを実装するという頭痛を完全に回避します。既に導入されている標準のHTTP認証を使用し続けています。既存のセキュリティパターンはすべて、常にと同じように機能し続けています。

以前にWebSocketsを実装したとき、これは私たちが尊敬しなければならなかったRBACの制限のために非常にひどくなりました。基本的に、接続されたクライアントにプッシュするデータと、クライアントがあるクラスターから別のクラスターに移動したときに発生する特権エスカレーションに本当に注意する必要がありました。

インフラストラクチャの互換性

WebSocket接続をブロックするコーポレートファイアウォールは、私たちの他の心配の1つでした。一部のユーザーはファイアウォールの背後にいますが、WebSocketを開くようにするというIT頭痛は必要ありません。

私たちの問題ではありません。特別なプロキシ構成や複雑なインフラストラクチャのセットアップは必要ありません。標準のロードバランサーの構成は、変更なしで正常に機能します。スタック全体は、いつものようにハミングを続けます。

運用上のシンプルさ

WebSocket接続を削除するサーバーの再起動について心配する必要はありません。管理または維持するための接続状態はありません。何かがうまくいかない場合（そして常に何かがうまくいかない）、標準のHTTPリクエストと応答を扱っているため、デバッグやトラブルシューティングがはるかに簡単になります。

CloudFlareをEdgeに使用します。つまり、既存の構成ルールとDDOS保護は変更を必要としませんでした。

クライアントの実装

クライアント側のコードは非常に単純なままです。 HTTPクライアントでは機能し、特別なWebSocketライブラリは必要ありません。さらに良いことに、再接続処理には基本的な再試行ロジックが付いています。クライアント全体の実装は、多くの場合、ほんの数行のコードになる可能性があります。

なぜElectricsQLをしないのですか？

解決策を探索している間、私たちは見ました ElectricsQl、Postgresデータをフロントエンドに同期させます。彼らは、WebSocketsを介した長い投票のために興味深いケースを作ります：

「HTTPプロトコルへの切り替えは、最初は回帰または奇妙な適合のように思えるかもしれません。WebSocketsはHTTPの上に構築されており、電動が提供するリアルタイムのデータストリームの種類を提供します。 。」

実際、リアルタイムの更新を処理するために極端な制御または低レベルのコンストラクトを必要としない場合は、実際にElectricsQLをお勧めします。これは、多くのエッジケースを処理し、優れた開発者エクスペリエンスを提供する強固な戦闘テストのソリューションです。

なぜ私たちが生の長い世論調査を選んだのか

メッセージ配信メカニズムは製品の中核部分です – それは単なる実装の詳細ではなく、私たちの仕事の中心です。そのライブラリがどんなに良くても、メッセージ配信がサードパーティライブラリで抽象化するような基本的なものを手に入れる余裕はありません。

私たちの特定のユースケースが必要です：

1. コア製品制御：メッセージ配信メカニズムを完全に制御する – それは単なるインフラストラクチャではなく、私たちの製品です
2. ゼロ外部依存関係：セルフホストのためにできるだけシンプルである必要がありました
3. 金属の近く：ポーリングメカニズムと接続処理を直接制御する
4. 最大制御：動的ポーリング間隔の実装を含む、実装のあらゆる側面を微調整する能力
5. シンプルさ：ユーザーがコードベースを簡単に理解して変更できるようにする

私たちにとって、シンプルなHTTPロングポーリングの実装で金属の近くにとどまることが正しい選択でした。ただし、このレベルの制御が必要ない場合、ElectricsQLは、重要な開発時間を節約できるより機能が豊富なソリューションを提供します。

アプリケーションレイヤーのベストプラクティス

長い世論調査を実施するとき、信頼できる操作を確保するためにいくつかの重要なプラクティスがあります。

必須のTTL実装

HTTP接続に時間（TTL）を実装する必要があります。これがなければ、必然的に接続リセットエラーにぶつかります。投票ロジックは、何があっても、常にこのTTL内に戻るはずです。

const getJobStatus = async (jobId: string, ttl = 60_000) => {
  const start = Date.now();
  
  
  while (Date.now() - start // polling logic here
  }
  
  return null; 
}

サーバー制限付きのクライアント制御可能なTTL

クライアントは目的のTTLを指定できる必要がありますが、サーバーは最大制限を実施する必要があります。

const MAX_TTL = 120_000; 

const getJobStatus = async (jobId: string, clientTtl: number) => {
  const ttl = Math.min(clientTtl, MAX_TTL);
  
}

インフラストラクチャを認識しているTTL設定

最大TTLは、インフラストラクチャスタック全体にわたる最小HTTP接続タイムアウトの下に留まる必要があります。

• アプリケーションサーバーのタイムアウト
• クライアントのタイムアウト
• ロードバランサーのタイムアウト
• エッジサーバーのタイムアウト
• プロキシタイムアウト

たとえば、Edgeサーバーに30秒のタイムアウトがある場合、Max TTLはこれで快適になるはずです。たとえば25秒。

賢明なデータベースポーリング間隔

実装に示されているように、データベース投票間の合理的な待機時間を含めます。 500ms間隔を使用します。

await new Promise(resolve => setTimeout(resolve, 500));

これにより、データベースを叩きながら、合理的に迅速な更新を提供します。

オプション：指数バックオフ

現在のシステムには実装されていませんが、より効率的なリソース使用量のために指数関数的バックオフを実装できます。

const getJobStatus = async (jobId: string, ttl = 60_000) => {
  const start = Date.now();
  let waitTime = 100; 
  
  while (Date.now() - start const result = await checkJob(jobId);
    
    if (result) return result;
    
    
    waitTime = Math.min(waitTime * 2, 2000);
    await new Promise(resolve => setTimeout(resolve, waitTime));
  }
  
  return null;
}

このアプローチは次のことを意味します：

• アクティブなリクエスト（すぐにデータを取得する可能性があるもの）が迅速に終了します
• 非アクティブな要求は、投票間隔を徐々に増やします
• システムリソースはより効率的に使用されます

WebSocketsのケース：ストーリーの反対側

私たちは長い世論調査を私たちのニーズのための素晴らしい解決策であることがわかりましたが、それが唯一の選択肢ではありません。 WebSocketsは本質的に悪くありません。彼らはただ多くの愛と注意を必要としています。

私たちが言及した課題は克服できないものではありません – 彼らはただ適切なエンジニアリングの注意を必要とします：

• 観察可能性：WebSocketsはよりステートフルなので、永続的な接続のために追加のログと監視を実装する必要があります。

• 認証：着信WebSocket接続の新しい認証メカニズムを実装する必要があります。

• インフラストラクチャー：ロードバランサーやファイアウォールなどのWebSocketをサポートするために、インフラストラクチャを構成する必要があります。

• 操作：接続のタイムアウトやエラーの処理など、WebSocket接続と再接続を管理する必要があります。

• クライアントの実装：再接続や状態管理の処理など、クライアント側のWebsocketライブラリを実装する必要があります。