khadhroony-solana-project/crates/ksp-onchain-transport-lib/unit_tests/executor.rs

// file: crates/ksp-onchain-transport-lib/unit_tests/executor.rs
// version: 1

fn pool_for_url(url: &str, request_timeout: std::time::Duration, max_retries: u32) -> crate::HttpTransportPool {
    let role = crate::HttpEndpointRoleSettings::new(
        crate::HttpRoleName::new("default"),
        true,
        std::vec![crate::HttpRequestKind::wildcard()],
        10,
        crate::HttpRoleLimits::new(
            std::option::Option::None,
            std::option::Option::None,
            std::option::Option::None,
            std::option::Option::Some(std::time::Duration::from_millis(1)),
        ),
    );
    let endpoint = crate::HttpEndpointSettings::new(
        "fixture",
        true,
        crate::HttpProviderName::new("fixture"),
        crate::HttpClusterName::new("local"),
        crate::HttpEndpointUrl::parse(url).expect("fixture URL must parse"),
        std::time::Duration::from_millis(100),
        request_timeout,
        std::option::Option::Some(1),
        std::vec![role],
    );
    let settings = crate::HttpTransportSettings::new(
        std::vec![endpoint],
        crate::HttpRetrySettings::new(max_retries, std::time::Duration::from_millis(1), std::time::Duration::from_millis(2)),
    );
    return crate::HttpTransportPool::new(settings).expect("fixture pool must build");
}

fn health_method() -> &'static crate::HttpRpcMethodDescriptor {
    return crate::find_http_rpc_method("getHealth").expect("getHealth descriptor must exist");
}

fn serve_rate_limit_then_success() -> (std::string::String, std::thread::JoinHandle<usize>) {
    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("fixture listener must bind");
    let address = listener.local_addr().expect("fixture listener address must resolve");
    let handle = std::thread::spawn(move || {
        let mut count = 0_usize;
        while count < 2 {
            let (mut stream, _) = listener.accept().expect("fixture server must accept request");
            let _ = read_request(&mut stream);
            let response = if count == 0 {
                "HTTP/1.1 429 Too Many Requests\r\nRetry-After: 0\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_owned()
            } else {
                let body = include_str!("../fixtures/http/get_health.success.json");
                format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", body.len(), body)
            };
            std::io::Write::write_all(&mut stream, response.as_bytes()).expect("fixture response must write");
            count = count.saturating_add(1);
        }
        return count;
    });
    return (format!("http://{address}"), handle);
}

fn serve_timeout() -> (std::string::String, std::thread::JoinHandle<()>) {
    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("fixture listener must bind");
    let address = listener.local_addr().expect("fixture listener address must resolve");
    let handle = std::thread::spawn(move || {
        let (mut stream, _) = listener.accept().expect("fixture server must accept request");
        let _ = read_request(&mut stream);
        std::thread::sleep(std::time::Duration::from_millis(100));
        return;
    });
    return (format!("http://{address}"), handle);
}

fn read_request(stream: &mut std::net::TcpStream) -> std::string::String {
    let mut bytes = std::vec::Vec::new();
    let mut buffer = [0_u8; 1024];
    loop {
        let count = std::io::Read::read(stream, &mut buffer).expect("fixture request must read");
        if count == 0 {
            break;
        }
        bytes.extend_from_slice(&buffer[..count]);
        if request_complete(bytes.as_slice()) {
            break;
        }
    }
    return std::string::String::from_utf8(bytes).expect("fixture request must be UTF-8");
}

fn request_complete(bytes: &[u8]) -> bool {
    let text = match std::str::from_utf8(bytes) {
        std::result::Result::Ok(text) => text,
        std::result::Result::Err(_) => return false,
    };
    let header_end = match text.find("\r\n\r\n") {
        std::option::Option::Some(value) => value,
        std::option::Option::None => return false,
    };
    let mut content_length = 0_usize;
    for line in text[..header_end].lines() {
        let (name, value) = match line.split_once(':') {
            std::option::Option::Some(parts) => parts,
            std::option::Option::None => continue,
        };
        if name.eq_ignore_ascii_case("content-length") {
            content_length = value.trim().parse::<usize>().expect("content length must parse");
        }
    }
    return bytes.len() >= header_end.saturating_add(4).saturating_add(content_length);
}

#[tokio::test(flavor = "current_thread")]
async fn executor_applies_retry_after_and_retries_http_429_for_retry_safe_method() {
    let (url, handle) = serve_rate_limit_then_success();
    let pool = pool_for_url(url.as_str(), std::time::Duration::from_millis(500), 1);
    let result = pool
        .execute_standard_rpc(&crate::HttpRoleName::new("default"), health_method(), std::vec::Vec::new())
        .await
        .expect("retry-safe request must recover from one 429");
    assert_eq!(result, serde_json::json!("ok"));
    assert_eq!(handle.join().expect("fixture server must join"), 2);
    let snapshot = pool.snapshot();
    assert_eq!(snapshot.endpoints()[0].roles()[0].rate_limit_count(), 1);
    assert_eq!(snapshot.endpoints()[0].roles()[0].success_count(), 1);
}

#[tokio::test(flavor = "current_thread")]
async fn executor_maps_reqwest_timeout_to_ksp_timeout_error() {
    let (url, handle) = serve_timeout();
    let pool = pool_for_url(url.as_str(), std::time::Duration::from_millis(20), 0);
    let error = pool
        .execute_standard_rpc(&crate::HttpRoleName::new("default"), health_method(), std::vec::Vec::new())
        .await
        .expect_err("timed out request must fail");
    assert_eq!(error.code(), crate::ERROR_CODE_TIMEOUT);
    handle.join().expect("fixture server must join");
}