待ち時間の測定

重大な免責事項: 本番データセンターのみこれらの待ち時間テストは、本番データセンター環境専用に設計されています。これらのテストをローカルマシンやコンシューマーインターネット接続で実行しないでください。ローカル帯域幅はソラナサブスクリプションを処理できず、実際のパフォーマンスを反映しない無意味な結果を生み出します。

同ロケーション要件: お使いのLaserStreamエンドポイント近くにデプロイする有意義な待ち時間の測定を行うには、選択したLaserStreamエンドポイントと同じ地域にテストインフラを配置する必要があります。ネットワーク距離が測定を支配するため、異なる大陸からテストすると、LaserStreamのパフォーマンスではなくネットワークの待ち時間が表示されます。

分散型ブロックチェーンシステムでの待ち時間の理解

ブロックチェーンストリーミングサービスを使用する場合、分散システムには普遍的な時計がないため、待ち時間の測定は複雑になります。単一サーバーへの往復時間を測定できる従来のシステムとは異なり、ブロックチェーンネットワークには複数のバリデータが関与しており、各バリデータは同じトランザクションを異なる時間に受信して処理します。 基本的な課題: Solanaのようなブロックチェーンには絶対時間の概念がありません。各バリデータノードは世界中で同じトランザクションを異なる時間に受信し、確認はクラスターのパーセンテージが合意に達することに依存しています。これにより、従来の意味での決定的な待ち時間測定が不可能になります。

コミットメントレベルと待ち時間の優先順位

Solanaは3つのコミットメントレベルを提供し、それぞれに異なる待ち時間特性があります:

Processed: 最速、単一バリデータの確認（約400ms）
Confirmed: 中程度、スーパーマジョリティーの確認（約2〜3秒）
Finalized: 最も遅い、ネットワーク全体の最終化（約15〜30秒）

待ち時間が重要なアプリケーションでは、通常、プロセス済みコミットメントが目標となります。このガイドのすべてのテストは、ほとんどの高頻度使用のケースが絶対的なファイナリティよりもスピードを優先するため、プロセス済みコミットメントレベルを使用しています。

待ち時間を測定する3つのアプローチ

1. 並列gRPCストリームの比較

最も信頼性の高い方法 - 同じデータソースに対して2つの独立したストリームを比較し、同一のイベントをどちらが先に受信するかを測定します。 利点:

時計同期の問題を排除
相対的なパフォーマンス比較を提供
サービス比較に最も正確

2. ローカルタイムスタンプとcreated_atの比較

中程度の信頼性 - LaserStreamサービスによってメッセージに埋め込まれたタイムスタンプと、システムがメッセージを受信した時刻の差を測定します。 制限事項:

LaserStreamが内部でメッセージを作成した時点のみを表す
LaserStreamへの上流遅延はキャプチャされない
真のエンドツーエンドの待ち時間に対してMethod 1よりも精度が低い

3. ブロックタイムスタンプ分析（非推奨）

非推奨 - Solanaのブロックタイムスタンプとローカルの受信時刻を比較します。 重要な制限:

ブロックタイムスタンプは秒単位の粒度しかない
Solanaは400msごとにブロックを生成
有用な情報はほとんど提供しない

セットアップ要件

地域的な同ロケーション

有意義な待ち時間測定を行うには、テストインフラをLaserStreamエンドポイントと同じデータセンターまたは地域にデプロイします。 利用可能なLaserStream地域:

ewr: ニューヨーク, US (東海岸)
pitt: ピッツバーグ, US (中部)
slc: ソルトレイクシティ, US (西海岸)
ams: アムステルダム, ヨーロッパ
fra: フランクフルト, ヨーロッパ
tyo: 東京, アジア
sgp: シンガポール, アジア

devnetテストには、次を使用します: 完全なセットアップ手順とエンドポイント選択ガイドラインについては、LaserStream gRPCドキュメントを参照してください。

Rust環境のセットアップ

すべての測定スクリプトはRustとCargoを使用します。基本設定:

# Install Rust
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

# Setup project
cargo new latency-testing
cd latency-testing

# Add dependencies to Cargo.toml
[dependencies]
# Async runtime
tokio = { version = "1", features = ["full"] }
# Futures utilities for StreamExt, SinkExt
futures = "0.3"
# Environment variable loading
dotenvy = "0.15"
# Logging
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] }
# Yellowstone gRPC client and protocol
yellowstone-grpc-client = "13.3.0"
yellowstone-grpc-proto = "12.6.0"
# For timestamp analysis
prost-types = "0.14"

認証情報を含むファイルを作成します:

YS_GRPC_URL=your-comparison-endpoint
YS_API_KEY=your-comparison-api-key
LS_GRPC_URL=your-laserstream-endpoint  
LS_API_KEY=your-helius-api-key

Helius APIキーはHelius Dashboardから入手できます。LaserStream devnetはすべてのプランで利用可能です。メインネットアクセスにはビジネスまたはプロフェッショナルプランが必要です。

方法1: 並列ストリーム比較

このスクリプトは、異なるgRPCエンドポイントへの2つの独立した接続を確立し、同じメッセージをどちらが先に受信するかを測定します。このアプローチは、相対的なタイミングを使用して時計同期の問題を排除します。

use std::collections::HashMap;
use std::time::SystemTime;

use dotenvy::dotenv;
use futures::{StreamExt, SinkExt};
use tokio::sync::mpsc;
use tracing::{debug, error, info};

use yellowstone_grpc_client::{ClientTlsConfig, GeyserGrpcClient};
use yellowstone_grpc_proto::prelude::{
    subscribe_update::UpdateOneof, CommitmentLevel, SubscribeRequest, 
    SubscribeRequestFilterBlocksMeta, SubscribeUpdate,
};

#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
enum Source {
    Yellowstone,
    Laserstream,
}

#[derive(Default, Debug)]
struct SlotTimings {
    ys_recv_ms: Option<i128>,
    ls_recv_ms: Option<i128>,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let _ = dotenv();
    tracing_subscriber::fmt().with_env_filter("info").init();

    let ys_url = std::env::var("YS_GRPC_URL").expect("YS_GRPC_URL env variable not set");
    let ls_url = std::env::var("LS_GRPC_URL").expect("LS_GRPC_URL env variable not set");
    let ys_api_key = std::env::var("YS_API_KEY").ok();
    let ls_api_key = std::env::var("LS_API_KEY").ok();

    let commitment = CommitmentLevel::Processed;
    info!(?commitment, "Starting latency comparison");

    // Establish both clients
    let mut ys_client = GeyserGrpcClient::build_from_shared(ys_url.clone())
        .expect("invalid YS url")
        .x_token(ys_api_key.clone())?
        .tls_config(ClientTlsConfig::new().with_native_roots())?
        .max_decoding_message_size(10 * 1024 * 1024)
        .connect()
        .await?;

    let mut ls_client = GeyserGrpcClient::build_from_shared(ls_url.clone())
        .expect("invalid LS url")
        .x_token(ls_api_key.clone())?
        .tls_config(ClientTlsConfig::new().with_native_roots())?
        .max_decoding_message_size(10 * 1024 * 1024)
        .connect()
        .await?;

    let (mut ys_tx, mut ys_rx) = ys_client.subscribe().await?;
    let (mut ls_tx, mut ls_rx) = ls_client.subscribe().await?;
    
    let subscribe_request = SubscribeRequest {
        blocks_meta: {
            let mut m = HashMap::<String, SubscribeRequestFilterBlocksMeta>::new();
            m.insert("all".to_string(), SubscribeRequestFilterBlocksMeta::default());
            m
        },
        commitment: Some(commitment as i32),
        ..Default::default()
    };

    ys_tx.send(subscribe_request.clone()).await?;
    ls_tx.send(subscribe_request).await?;

    let (agg_tx, mut agg_rx) = mpsc::unbounded_channel::<(Source, u64, i128)>();

    // Spawn task for Yellowstone stream
    {
        let agg_tx = agg_tx.clone();
        tokio::spawn(async move {
            while let Some(update_res) = ys_rx.next().await {
                match update_res {
                    Ok(update) => handle_update(Source::Yellowstone, update, &agg_tx).await,
                    Err(e) => {
                        error!(target: "ys", "stream error: {:?}", e);
                        break;
                    }
                }
            }
        });
    }

    // Spawn task for Laserstream stream
    {
        let agg_tx = agg_tx.clone();
        tokio::spawn(async move {
            while let Some(update_res) = ls_rx.next().await {
                match update_res {
                    Ok(update) => handle_update(Source::Laserstream, update, &agg_tx).await,
                    Err(e) => {
                        error!(target: "ls", "stream error: {:?}", e);
                        break;
                    }
                }
            }
        });
    }

    // Aggregator – collect latencies per slot and print once we have both sources
    let mut timings: HashMap<u64, SlotTimings> = HashMap::new();
    let mut deltas: Vec<i128> = Vec::new();
    let mut count = 0;

    println!("slot,ys_recv_ms,ls_recv_ms,delta_ms");

    while let Some((source, slot, latency_ms)) = agg_rx.recv().await {
        let entry = timings.entry(slot).or_default();
        match source {
            Source::Yellowstone => entry.ys_recv_ms = Some(latency_ms),
            Source::Laserstream => entry.ls_recv_ms = Some(latency_ms),
        }

        if let (Some(ys), Some(ls)) = (entry.ys_recv_ms, entry.ls_recv_ms) {
            let delta = ys - ls; // positive => YS arrived later
            println!("{slot},{ys},{ls},{delta}");
            
            deltas.push(delta);
            count += 1;
            
            if count % 100 == 0 {
                print_statistics(&deltas, count);
            }
            
            timings.remove(&slot);
        }
    }

    Ok(())
}

async fn handle_update(
    source: Source,
    update: SubscribeUpdate,
    agg_tx: &mpsc::UnboundedSender<(Source, u64, i128)>,
) {
    if let Some(UpdateOneof::BlockMeta(block_meta)) = update.update_oneof {
        let slot = block_meta.slot;
        let recv_ms = system_time_to_millis(SystemTime::now());
        debug!(?source, slot, recv_ms, "BlockMeta received");
        let _ = agg_tx.send((source, slot, recv_ms));
    }
}

fn print_statistics(deltas: &[i128], count: usize) {
    if deltas.is_empty() {
        return;
    }
    
    let mut sorted_deltas = deltas.to_vec();
    sorted_deltas.sort();
    
    let median = if sorted_deltas.len() % 2 == 0 {
        let mid = sorted_deltas.len() / 2;
        (sorted_deltas[mid - 1] + sorted_deltas[mid]) / 2
    } else {
        sorted_deltas[sorted_deltas.len() / 2]
    };
    
    let min = *sorted_deltas.first().unwrap();
    let max = *sorted_deltas.last().unwrap();
    let sum: i128 = sorted_deltas.iter().sum();
    let mean = sum / sorted_deltas.len() as i128;
    
    let p25_idx = (sorted_deltas.len() as f64 * 0.25) as usize;
    let p75_idx = (sorted_deltas.len() as f64 * 0.75) as usize;
    let p95_idx = (sorted_deltas.len() as f64 * 0.95) as usize;
    
    let p25 = sorted_deltas[p25_idx.min(sorted_deltas.len() - 1)];
    let p75 = sorted_deltas[p75_idx.min(sorted_deltas.len() - 1)];
    let p95 = sorted_deltas[p95_idx.min(sorted_deltas.len() - 1)];
    
    eprintln!("--- Statistics after {} slots ---", count);
    eprintln!("Delta (YS - LS) in milliseconds:");
    eprintln!("  Min: {}, Max: {}", min, max);
    eprintln!("  Mean: {}, Median: {}", mean, median);
    eprintln!("  P25: {}, P75: {}, P95: {}", p25, p75, p95);
    eprintln!("  Positive deltas (YS slower): {}/{} ({:.1}%)", 
              sorted_deltas.iter().filter(|&&x| x > 0).count(),
              sorted_deltas.len(),
              sorted_deltas.iter().filter(|&&x| x > 0).count() as f64 / sorted_deltas.len() as f64 * 100.0);
    eprintln!("---");
}

fn system_time_to_millis(st: SystemTime) -> i128 {
    st.duration_since(SystemTime::UNIX_EPOCH)
        .unwrap()
        .as_millis() as i128
}

これが測定するもの: 2つのストリーミングサービス間の相対的なパフォーマンスの違い。デルタは、どのサービスが同じスロット情報を最初に提供するかを示します。 主要な指標:

正のデルタ: 最初のサービス（YS）が2番目のサービス（LS）より遅い - LaserStreamの方が速い
負のデルタ: 最初のサービス（YS）が2番目のサービス（LS）より速い - LaserStreamの方が遅い
平均/中央値: 平均パフォーマンスの違い
P95: 95パーセンタイルの待ち時間差

テストの実行:

cargo run --bin latency-comparison

サンプル出力:

slot,ys_recv_ms,ls_recv_ms,delta_ms
352416939,1752168399141,1752168399140,1
352416940,1752168399526,1752168399512,14
352416941,1752168399890,1752168399877,13

出力はリアルタイムの待ち時間差と定期的な統計を表示します。正の平均デルタは、2番目のサービス（LaserStream）が一貫してデータをより速く提供していることを示します。

方法2: 作成タイムスタンプ分析

このアプローチは、メッセージに埋め込まれたタイムスタンプをローカルシステム時間と比較します。

use std::time::{Duration, SystemTime};
use dotenvy::dotenv;
use tracing::{debug, error, info};
use yellowstone_grpc_proto::prost_types::Timestamp;
use futures::StreamExt;
use futures::SinkExt;
use std::collections::HashMap;

use yellowstone_grpc_client::{ClientTlsConfig, GeyserGrpcClient};
use yellowstone_grpc_proto::prelude::{
    subscribe_update::UpdateOneof, CommitmentLevel, SubscribeRequest,
    SubscribeRequestFilterTransactions, SubscribeUpdate,
};

const ACCOUNTS_INCLUDE: &[&str] = &["BB5dnY55FXS1e1NXqZDwCzgdYJdMCj3B92PU6Q5Fb6DT"];
const COMMITMENT_LEVEL: &str = "processed";

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let _ = dotenv();
    tracing_subscriber::fmt().with_env_filter("info").init();

    let grpc_url = std::env::var("YS_GRPC_URL").expect("GRPC_URL env variable not set");
    let api_key = std::env::var("YS_API_KEY").ok();

    info!("Connecting to {} …", grpc_url);
    debug!(accounts_include = ?ACCOUNTS_INCLUDE, "Subscribing with accountsInclude filter");

    let mut client = GeyserGrpcClient::build_from_shared(grpc_url.clone())
        .expect("invalid URL")
        .x_token(api_key.clone())?
        .tls_config(ClientTlsConfig::new().with_native_roots())?
        .connect()
        .await?;

    let (mut subscribe_tx, mut subscribe_rx) = client.subscribe().await?;

    let mut tx_filter_map = HashMap::new();
    tx_filter_map.insert(
        "latency".to_string(),
        SubscribeRequestFilterTransactions {
            account_include: ACCOUNTS_INCLUDE.iter().map(|s| s.to_string()).collect(),
            vote: Some(false),
            failed: Some(false),
            ..Default::default()
        },
    );

    subscribe_tx
        .send(SubscribeRequest {
            transactions: tx_filter_map,
            commitment: Some(CommitmentLevel::Processed as i32),
            ..Default::default()
        })
        .await?;

    let mut latencies: Vec<i64> = Vec::new();
    let mut count = 0;

    while let Some(update_res) = subscribe_rx.next().await {
        match update_res {
            Ok(update) => {
                if let Some(UpdateOneof::Transaction(tx)) = update.update_oneof {
                    let recv_time = SystemTime::now();
                    
                    if let Some(created_at) = update.created_at {
                        let created_time = SystemTime::UNIX_EPOCH + Duration::new(
                            created_at.seconds as u64,
                            created_at.nanos as u32,
                        );
                        
                        if let Ok(latency) = recv_time.duration_since(created_time) {
                            let latency_ms = latency.as_millis() as i64;
                            latencies.push(latency_ms);
                            count += 1;
                            
                            println!("Transaction latency: {}ms", latency_ms);
                            
                            if count % 100 == 0 {
                                print_statistics(&latencies, count);
                            }
                        }
                    }
                }
            }
            Err(e) => {
                error!("Stream error: {:?}", e);
                break;
            }
        }
    }

    Ok(())
}

fn print_statistics(latencies: &[i64], count: usize) {
    if latencies.is_empty() {
        return;
    }
    
    let mut sorted = latencies.to_vec();
    sorted.sort();
    
    let median = if sorted.len() % 2 == 0 {
        let mid = sorted.len() / 2;
        (sorted[mid - 1] + sorted[mid]) / 2
    } else {
        sorted[sorted.len() / 2]
    };
    
    let min = *sorted.first().unwrap();
    let max = *sorted.last().unwrap();
    let sum: i64 = sorted.iter().sum();
    let mean = sum / sorted.len() as i64;
    
    let p95_idx = (sorted.len() as f64 * 0.95) as usize;
    let p95 = sorted[p95_idx.min(sorted.len() - 1)];
    
    eprintln!("--- Statistics after {} transactions ---", count);
    eprintln!("Latency (created_at to receive) in milliseconds:");
    eprintln!("  Min: {}, Max: {}", min, max);
    eprintln!("  Mean: {}, Median: {}", mean, median);
    eprintln!("  P95: {}", p95);
    eprintln!("---");
}

重要な制限: この方法は、LaserStreamがメッセージを作成したときから受信したときまでの時間のみを測定します。ブロックチェーンイベントとLaserStream処理の間の上流遅延は考慮されません。 テストの実行:

cargo run --bin timestamp-analysis

サンプル出力:

Transaction latency: 45ms
Transaction latency: 52ms
Transaction latency: 38ms
--- Statistics after 100 transactions ---
Latency (created_at to receive) in milliseconds:
  Min: 28, Max: 89
  Mean: 47, Median: 45
  P95: 72
---

この方法は、LaserStreamとアプリケーション間のネットワークおよび処理の待ち時間に関する洞察を提供しますが、Method 1と組み合わせて使用することで包括的な分析が可能です。

待ち時間テストのベストプラクティス

重要な原則

同ロケーション: テストをLaserStreamエンドポイントと同じ地域にデプロイしてネットワーク待ち時間を最小化
複数の方法: 並列ストリーム比較（方法1）を主要な指標として使用し、タイムスタンプ分析で補完
長期監視: テストを長期間にわたって実行し、異なるネットワーク条件やブロックチェーンの混雑を把握
統計分析: 平均だけでなくパーセンタイル（P95, P99）に焦点を当ててテール待ち時間を把握

結果の解釈

ベースラインの確立: 通常の条件下でのベースラインパフォーマンスを確立するために最低1時間のテストを実施
パターンの特定: 待ち時間のスパイクのパターンを探し、ブロックチェーンの高いアクティビティやネットワークの混雑と関連があるか確認
パーセンタイルの比較: ユーザーエクスペリエンスにおいては平均待ち時間よりもP95待ち時間が重要であることが多い
一貫性の監視: 絶対最小待ち時間よりも一貫したパフォーマンスがより価値があることが多い

ブロックチェーンの待ち時間は、ネットワークの合意要件のため本質的に変動します。絶対値よりも相対的なパフォーマンスの違いと一貫性に焦点を当てましょう。

ここから始める

リアルタイムデータ

低レイテンシ取引

履歴データとウォレットデータ

RPC メソッドガイド

本番環境への対応

Orb の使用

その他のハウツー

分散型ブロックチェーンシステムでの待ち時間の理解

コミットメントレベルと待ち時間の優先順位

待ち時間を測定する3つのアプローチ

1. 並列gRPCストリームの比較

2. ローカルタイムスタンプとcreated_atの比較

3. ブロックタイムスタンプ分析（非推奨）

セットアップ要件

地域的な同ロケーション

Rust環境のセットアップ

方法1: 並列ストリーム比較

方法2: 作成タイムスタンプ分析

待ち時間テストのベストプラクティス

重要な原則

結果の解釈

​分散型ブロックチェーンシステムでの待ち時間の理解

​コミットメントレベルと待ち時間の優先順位

​待ち時間を測定する3つのアプローチ

​1. 並列gRPCストリームの比較

​2. ローカルタイムスタンプとcreated_atの比較

​3. ブロックタイムスタンプ分析（非推奨）

​セットアップ要件

​地域的な同ロケーション

​Rust環境のセットアップ

​方法1: 並列ストリーム比較

​方法2: 作成タイムスタンプ分析

​待ち時間テストのベストプラクティス

​重要な原則

​結果の解釈

分散型ブロックチェーンシステムでの待ち時間の理解

コミットメントレベルと待ち時間の優先順位

待ち時間を測定する3つのアプローチ

1. 並列gRPCストリームの比較

2. ローカルタイムスタンプとcreated_atの比較

3. ブロックタイムスタンプ分析（非推奨）

セットアップ要件

地域的な同ロケーション

Rust環境のセットアップ

方法1: 並列ストリーム比較

方法2: 作成タイムスタンプ分析

待ち時間テストのベストプラクティス

重要な原則

結果の解釈