Repository: MystenLabs/sui
File: crates/sui-e2e-tests/tests/authenticated_events_api_tests.rs (Rust)

Code before:
        vec![],
        vec![val, sz],
    );
    let gas_object = test_cluster
        .wallet
        .get_one_gas_object_owned_by_address(sender)
        .await
        .unwrap()
        .unwrap();
    let tx_data = TransactionData::new(
        sui_types::transaction::TransactionKind::ProgrammableTransaction(ptb.finish()),
        sender,
        gas_object,
        50_000_000_000,
        rgp,
    );
    test_cluster.sign_and_execute_transaction(&tx_data).await
}

async fn query_authenticated_events(
    rpc_url: &str,
    stream_id: &str,
    start_checkpoint: u64,
    page_size: Option<u32>,
) -> Result<
    sui_rpc_api::grpc::alpha::event_service_proto::ListAuthenticatedEventsResponse,
    tonic::Status,
> {
    let mut client = EventServiceClient::connect(rpc_url.to_owned())
        .await
        .unwrap();

    let mut req = ListAuthenticatedEventsRequest::default();
    req.stream_id = Some(stream_id.to_string());
    req.start_checkpoint = Some(start_checkpoint);
    req.page_size = page_size;
    req.page_token = None;

    client
        .list_authenticated_events(req)
        .await
        .map(|r| r.into_inner())
}

async fn list_authenticated_events(
    rpc_url: &str,
    stream_id: &str,
    start_checkpoint: u64,
    page_size: Option<u32>,
) -> Vec<AuthenticatedEvent> {
    let mut event_client = EventServiceClient::connect(rpc_url.to_owned())

Review comment (bug):
It seems like all gRPC clients would be equally susceptible to this bug. Should we generalize this solution?

Code after:
        move_core_types::identifier::Identifier::new("emit_large").unwrap(),
        vec![],
        vec![val, sz],
    );
    let gas_object = test_cluster
        .wallet
        .get_one_gas_object_owned_by_address(sender)
        .await
        .unwrap()
        .unwrap();
    let tx_data = TransactionData::new(
        sui_types::transaction::TransactionKind::ProgrammableTransaction(ptb.finish()),
        sender,
        gas_object,
        50_000_000_000,
        rgp,
    );
    test_cluster.sign_and_execute_transaction(&tx_data).await
}

/// Connect to the EventServiceClient with timeout and retry logic.
///
/// gRPC connection establishment can hang indefinitely if the remote peer is unable to complete
/// connection establishment. This helper ensures we always have bounded connection times and
/// can retry on transient failures.
async fn connect_event_service_client(
    rpc_url: &str,
) -> EventServiceClient<tonic::transport::Channel> {
    const MAX_RETRIES: u32 = 10;
    const CONNECT_TIMEOUT: Duration = Duration::from_secs(1);

    for attempt in 0..MAX_RETRIES {
        match tokio::time::timeout(
            CONNECT_TIMEOUT,
            EventServiceClient::connect(rpc_url.to_owned()),
        )
        .await
        {
            Ok(Ok(client)) => return client,
            Ok(Err(e)) if attempt + 1 < MAX_RETRIES => {
                tokio::time::sleep(Duration::from_millis(100)).await;
                continue;
            }
            Ok(Err(e)) => panic!("failed to connect after {MAX_RETRIES} attempts: {e}"),
            Err(_) if attempt + 1 < MAX_RETRIES => {
                tokio::time::sleep(Duration::from_millis(100)).await;
                continue;
            }
            Err(_) => panic!("connection timed out after {MAX_RETRIES} attempts"),
        }
    }