// file: crates/ksp-onchain-transport-lib/unit_tests/grpc_unary.rs // version: 3 #[derive(Clone, Default)] struct FixtureGeyser; #[allow(clippy::implicit_return)] // tonic::async_trait generates async wrapper tails outside the authored fixture bodies. #[tonic::async_trait] impl yellowstone_grpc_proto::geyser::geyser_server::Geyser for FixtureGeyser { type SubscribeStream = std::pin::Pin< std::boxed::Box> + Send + 'static>, >; async fn subscribe( &self, _request: tonic::Request>, ) -> std::result::Result, tonic::Status> { return std::result::Result::Err(tonic::Status::unimplemented("streaming is outside the pre.003 fixture")); } type SubscribeDeshredStream = std::pin::Pin< std::boxed::Box< dyn futures_util::Stream> + Send + 'static, >, >; async fn subscribe_deshred( &self, _request: tonic::Request>, ) -> std::result::Result, tonic::Status> { return std::result::Result::Err(tonic::Status::unimplemented("deshred is outside KSP 0.2.9")); } async fn subscribe_replay_info( &self, request: tonic::Request, ) -> std::result::Result, tonic::Status> { if let std::result::Result::Err(error) = verify_fixture_metadata(request.metadata()) { return std::result::Result::Err(error); } return std::result::Result::Ok(tonic::Response::new(yellowstone_grpc_proto::geyser::SubscribeReplayInfoResponse { first_available: std::option::Option::Some(100), })); } async fn ping( &self, request: tonic::Request, ) -> std::result::Result, tonic::Status> { if let std::result::Result::Err(error) = verify_fixture_metadata(request.metadata()) { return std::result::Result::Err(error); } let count = request.into_inner().count; if count == -999 { return std::result::Result::Err(tonic::Status::permission_denied("GRPC-REMOTE-SECRET-CANARY")); } return std::result::Result::Ok(tonic::Response::new(yellowstone_grpc_proto::geyser::PongResponse { count })); } async fn get_latest_blockhash( &self, request: tonic::Request, ) -> std::result::Result, tonic::Status> { if let std::result::Result::Err(error) = verify_fixture_metadata(request.metadata()) { return std::result::Result::Err(error); } if let std::result::Result::Err(error) = require_commitment(request.get_ref().commitment, yellowstone_grpc_proto::geyser::CommitmentLevel::Confirmed) { return std::result::Result::Err(error); } return std::result::Result::Ok(tonic::Response::new(yellowstone_grpc_proto::geyser::GetLatestBlockhashResponse { slot: 101, blockhash: "fixture-blockhash".to_owned(), last_valid_block_height: 999, })); } async fn get_block_height( &self, request: tonic::Request, ) -> std::result::Result, tonic::Status> { if let std::result::Result::Err(error) = verify_fixture_metadata(request.metadata()) { return std::result::Result::Err(error); } if let std::result::Result::Err(error) = require_commitment(request.get_ref().commitment, yellowstone_grpc_proto::geyser::CommitmentLevel::Confirmed) { return std::result::Result::Err(error); } return std::result::Result::Ok(tonic::Response::new(yellowstone_grpc_proto::geyser::GetBlockHeightResponse { block_height: 202 })); } async fn get_slot( &self, request: tonic::Request, ) -> std::result::Result, tonic::Status> { if let std::result::Result::Err(error) = verify_fixture_metadata(request.metadata()) { return std::result::Result::Err(error); } if request.get_ref().commitment == std::option::Option::Some(yellowstone_grpc_proto::geyser::CommitmentLevel::Processed as i32) { tokio::time::sleep(std::time::Duration::from_millis(50)).await; } else if let std::result::Result::Err(error) = require_commitment(request.get_ref().commitment, yellowstone_grpc_proto::geyser::CommitmentLevel::Confirmed) { return std::result::Result::Err(error); } return std::result::Result::Ok(tonic::Response::new(yellowstone_grpc_proto::geyser::GetSlotResponse { slot: 303 })); } async fn is_blockhash_valid( &self, request: tonic::Request, ) -> std::result::Result, tonic::Status> { if let std::result::Result::Err(error) = verify_fixture_metadata(request.metadata()) { return std::result::Result::Err(error); } if let std::result::Result::Err(error) = require_commitment(request.get_ref().commitment, yellowstone_grpc_proto::geyser::CommitmentLevel::Confirmed) { return std::result::Result::Err(error); } if request.get_ref().blockhash != "fixture-blockhash" { return std::result::Result::Err(tonic::Status::invalid_argument("unexpected fixture blockhash")); } return std::result::Result::Ok(tonic::Response::new(yellowstone_grpc_proto::geyser::IsBlockhashValidResponse { slot: 404, valid: true })); } async fn get_version( &self, request: tonic::Request, ) -> std::result::Result, tonic::Status> { if let std::result::Result::Err(error) = verify_fixture_metadata(request.metadata()) { return std::result::Result::Err(error); } return std::result::Result::Ok(tonic::Response::new(yellowstone_grpc_proto::geyser::GetVersionResponse { version: "fixture-yellowstone-12.6".to_owned(), })); } } struct FixtureServer { endpoint_url: std::string::String, shutdown: std::option::Option>, task: tokio::task::JoinHandle<()>, } impl FixtureServer { async fn start() -> Self { let bind_address: std::net::SocketAddr = "127.0.0.1:0".parse().expect("fixture bind address must parse"); let incoming = tonic::transport::server::TcpIncoming::bind(bind_address).expect("fixture gRPC listener must bind"); let local_address = incoming.local_addr().expect("fixture gRPC listener must expose local address"); let (shutdown, shutdown_receiver) = tokio::sync::oneshot::channel(); let task = tokio::spawn(async move { let service = yellowstone_grpc_proto::geyser::geyser_server::GeyserServer::new(FixtureGeyser); let result = tonic::transport::Server::builder() .serve_with_incoming_shutdown(service, incoming, async move { let _ = shutdown_receiver.await; }) .await; assert!(result.is_ok()); }); return Self { endpoint_url: format!("http://{local_address}"), shutdown: std::option::Option::Some(shutdown), task }; } async fn stop(mut self) { if let std::option::Option::Some(shutdown) = self.shutdown.take() { let _ = shutdown.send(()); } let result = self.task.await; assert!(result.is_ok()); } } fn verify_fixture_metadata(metadata: &tonic::metadata::MetadataMap) -> std::result::Result<(), tonic::Status> { let public = metadata.get("x-ksp-public").and_then(|value| return value.to_str().ok()); let secret = metadata.get("x-ksp-token").and_then(|value| return value.to_str().ok()); if public != std::option::Option::Some("fixture-public") || secret != std::option::Option::Some("GRPC-SECRET-CANARY") { return std::result::Result::Err(tonic::Status::unauthenticated("fixture metadata mismatch")); } return std::result::Result::Ok(()); } fn require_commitment(actual: std::option::Option, expected: yellowstone_grpc_proto::geyser::CommitmentLevel) -> std::result::Result<(), tonic::Status> { if actual != std::option::Option::Some(expected as i32) { return std::result::Result::Err(tonic::Status::invalid_argument("fixture commitment mismatch")); } return std::result::Result::Ok(()); } fn fixture_settings(url: &str, unary_timeout: std::time::Duration) -> crate::YellowstoneGrpcEndpointSettings { let defaults = crate::YellowstoneGrpcSessionSettings::default(); let session = crate::YellowstoneGrpcSessionSettings::new( defaults.connect_timeout(), unary_timeout, defaults.close_timeout(), defaults.reconnect().clone(), defaults.request_channel_capacity(), defaults.update_channel_capacity(), defaults.max_inbound_message_size_bytes(), defaults.max_outbound_message_size_bytes(), ); let metadata = std::vec![ crate::YellowstoneGrpcMetadataEntry::public("x-ksp-public", "fixture-public").expect("fixture public metadata must be valid"), crate::YellowstoneGrpcMetadataEntry::secret("x-ksp-token", "GRPC-SECRET-CANARY").expect("fixture secret metadata must be valid"), ]; return crate::YellowstoneGrpcEndpointSettings::new( "fixture", true, crate::YellowstoneGrpcProviderName::new("fixture-provider"), crate::YellowstoneGrpcClusterName::new("devnet"), crate::YellowstoneGrpcEndpointUrl::parse(url).expect("fixture URL must parse"), session, ) .with_metadata(metadata) .expect("fixture metadata settings must validate"); } #[tokio::test(flavor = "current_thread")] async fn yellowstone_unary_fixture_covers_all_seven_standard_methods_and_metadata() { let server = FixtureServer::start().await; let settings = fixture_settings(server.endpoint_url.as_str(), std::time::Duration::from_secs(1)); let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect"); let client = channel.standard_unary_client(); let replay = client.subscribe_replay_info().await.expect("ReplayInfo fixture must succeed"); assert_eq!(replay.first_available(), std::option::Option::Some(100)); let pong = client.ping(7).await.expect("Ping fixture must succeed"); assert_eq!(pong.count(), 7); let latest = client .get_latest_blockhash(std::option::Option::Some(crate::SolanaCommitment::Confirmed)) .await .expect("latest blockhash fixture must succeed"); assert_eq!(latest.slot(), 101); assert_eq!(latest.blockhash(), "fixture-blockhash"); assert_eq!(latest.last_valid_block_height(), 999); let height = client.get_block_height(std::option::Option::Some(crate::SolanaCommitment::Confirmed)).await.expect("block height fixture must succeed"); assert_eq!(height.block_height(), 202); let slot = client.get_slot(std::option::Option::Some(crate::SolanaCommitment::Confirmed)).await.expect("slot fixture must succeed"); assert_eq!(slot.slot(), 303); let validity = client .is_blockhash_valid("fixture-blockhash", std::option::Option::Some(crate::SolanaCommitment::Confirmed)) .await .expect("blockhash validity fixture must succeed"); assert_eq!(validity.slot(), 404); assert!(validity.valid()); let version = client.get_version().await.expect("version fixture must succeed"); assert_eq!(version.version(), "fixture-yellowstone-12.6"); let rendered = format!("{client:?} {channel:?}"); assert!(!rendered.contains("GRPC-SECRET-CANARY")); assert!(!rendered.contains(server.endpoint_url.as_str())); server.stop().await; } #[tokio::test(flavor = "current_thread")] async fn yellowstone_remote_status_does_not_copy_message_details_or_metadata() { let server = FixtureServer::start().await; let settings = fixture_settings(server.endpoint_url.as_str(), std::time::Duration::from_secs(1)); let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect"); let error = channel.standard_unary_client().ping(-999).await.expect_err("fixture must return remote status"); assert_eq!(error.code(), crate::ERROR_CODE_GRPC_STATUS); let rendered = format!("{error:?}"); assert!(!rendered.contains("GRPC-REMOTE-SECRET-CANARY")); assert!(!rendered.contains("GRPC-SECRET-CANARY")); assert!(!rendered.contains(server.endpoint_url.as_str())); server.stop().await; } #[tokio::test(flavor = "current_thread")] async fn yellowstone_unary_timeout_is_bounded_and_safe() { let server = FixtureServer::start().await; let settings = fixture_settings(server.endpoint_url.as_str(), std::time::Duration::from_millis(5)); let channel = crate::YellowstoneGrpcChannel::connect(&settings).await.expect("fixture channel must connect"); let error = channel .standard_unary_client() .get_slot(std::option::Option::Some(crate::SolanaCommitment::Processed)) .await .expect_err("delayed fixture must exceed unary timeout"); assert_eq!(error.code(), crate::ERROR_CODE_TIMEOUT); let rendered = format!("{error:?}"); assert!(!rendered.contains("GRPC-SECRET-CANARY")); assert!(!rendered.contains(server.endpoint_url.as_str())); server.stop().await; } #[tokio::test(flavor = "current_thread")] async fn yellowstone_invalid_blockhash_is_rejected_before_io() { let settings = fixture_settings("http://127.0.0.1:1", std::time::Duration::from_secs(1)); let channel = crate::YellowstoneGrpcChannel::prepare(&settings).expect("lazy fixture channel must prepare"); let error = channel .standard_unary_client() .is_blockhash_valid(" ", std::option::Option::None) .await .expect_err("invalid blockhash must be rejected locally"); assert_eq!(error.code(), crate::ERROR_CODE_INVALID_RPC_PARAMETERS); }