drive_abci/query/

service.rs

use crate::error::query::QueryError;
use crate::error::Error;
use crate::metrics::{abci_response_code_metric_label, query_duration_metric};
use crate::platform_types::platform::Platform;
use crate::platform_types::platform_state::PlatformState;
use crate::platform_types::platform_state::PlatformStateV0Methods;
use crate::query::QueryValidationResult;
use crate::rpc::core::DefaultCoreRPC;
use crate::utils::spawn_blocking_task_with_name_if_supported;
use async_trait::async_trait;
use dapi_grpc::drive::v0::drive_internal_server::DriveInternal;
use dapi_grpc::drive::v0::{GetProofsRequest, GetProofsResponse};
use dapi_grpc::platform::v0::platform_server::Platform as PlatformService;
use dapi_grpc::platform::v0::{
    BroadcastStateTransitionRequest, BroadcastStateTransitionResponse, GetAddressInfoRequest,
    GetAddressInfoResponse, GetAddressesBranchStateRequest, GetAddressesBranchStateResponse,
    GetAddressesInfosRequest, GetAddressesInfosResponse, GetAddressesTrunkStateRequest,
    GetAddressesTrunkStateResponse, GetConsensusParamsRequest, GetConsensusParamsResponse,
    GetContestedResourceIdentityVotesRequest, GetContestedResourceIdentityVotesResponse,
    GetContestedResourceVoteStateRequest, GetContestedResourceVoteStateResponse,
    GetContestedResourceVotersForIdentityRequest, GetContestedResourceVotersForIdentityResponse,
    GetContestedResourcesRequest, GetContestedResourcesResponse, GetCurrentQuorumsInfoRequest,
    GetCurrentQuorumsInfoResponse, GetDataContractHistoryRequest, GetDataContractHistoryResponse,
    GetDataContractRequest, GetDataContractResponse, GetDataContractsRequest,
    GetDataContractsResponse, GetDocumentHistoryRequest, GetDocumentHistoryResponse,
    GetDocumentsRequest, GetDocumentsResponse, GetEpochsInfoRequest, GetEpochsInfoResponse,
    GetEvonodesProposedEpochBlocksByIdsRequest, GetEvonodesProposedEpochBlocksByRangeRequest,
    GetEvonodesProposedEpochBlocksResponse, GetFinalizedEpochInfosRequest,
    GetFinalizedEpochInfosResponse, GetGroupActionSignersRequest, GetGroupActionSignersResponse,
    GetGroupActionsRequest, GetGroupActionsResponse, GetGroupInfoRequest, GetGroupInfoResponse,
    GetGroupInfosRequest, GetGroupInfosResponse, GetIdentitiesBalancesRequest,
    GetIdentitiesBalancesResponse, GetIdentitiesContractKeysRequest,
    GetIdentitiesContractKeysResponse, GetIdentitiesTokenBalancesRequest,
    GetIdentitiesTokenBalancesResponse, GetIdentitiesTokenInfosRequest,
    GetIdentitiesTokenInfosResponse, GetIdentityBalanceAndRevisionRequest,
    GetIdentityBalanceAndRevisionResponse, GetIdentityBalanceRequest, GetIdentityBalanceResponse,
    GetIdentityByNonUniquePublicKeyHashRequest, GetIdentityByNonUniquePublicKeyHashResponse,
    GetIdentityByPublicKeyHashRequest, GetIdentityByPublicKeyHashResponse,
    GetIdentityContractNonceRequest, GetIdentityContractNonceResponse, GetIdentityKeysRequest,
    GetIdentityKeysResponse, GetIdentityNonceRequest, GetIdentityNonceResponse, GetIdentityRequest,
    GetIdentityResponse, GetIdentityTokenBalancesRequest, GetIdentityTokenBalancesResponse,
    GetIdentityTokenInfosRequest, GetIdentityTokenInfosResponse,
    GetMostRecentShieldedAnchorRequest, GetMostRecentShieldedAnchorResponse,
    GetNullifiersBranchStateRequest, GetNullifiersBranchStateResponse,
    GetNullifiersTrunkStateRequest, GetNullifiersTrunkStateResponse, GetPathElementsRequest,
    GetPathElementsResponse, GetPrefundedSpecializedBalanceRequest,
    GetPrefundedSpecializedBalanceResponse, GetProtocolVersionUpgradeStateRequest,
    GetProtocolVersionUpgradeStateResponse, GetProtocolVersionUpgradeVoteStatusRequest,
    GetProtocolVersionUpgradeVoteStatusResponse, GetRecentAddressBalanceChangesRequest,
    GetRecentAddressBalanceChangesResponse, GetRecentCompactedAddressBalanceChangesRequest,
    GetRecentCompactedAddressBalanceChangesResponse, GetRecentCompactedNullifierChangesRequest,
    GetRecentCompactedNullifierChangesResponse, GetRecentNullifierChangesRequest,
    GetRecentNullifierChangesResponse, GetShieldedAnchorsRequest, GetShieldedAnchorsResponse,
    GetShieldedEncryptedNotesRequest, GetShieldedEncryptedNotesResponse,
    GetShieldedNotesCountRequest, GetShieldedNotesCountResponse, GetShieldedNullifiersRequest,
    GetShieldedNullifiersResponse, GetShieldedPoolStateRequest, GetShieldedPoolStateResponse,
    GetStatusRequest, GetStatusResponse, GetTokenContractInfoRequest, GetTokenContractInfoResponse,
    GetTokenDirectPurchasePricesRequest, GetTokenDirectPurchasePricesResponse,
    GetTokenPerpetualDistributionLastClaimRequest, GetTokenPerpetualDistributionLastClaimResponse,
    GetTokenPreProgrammedDistributionsRequest, GetTokenPreProgrammedDistributionsResponse,
    GetTokenStatusesRequest, GetTokenStatusesResponse, GetTokenTotalSupplyRequest,
    GetTokenTotalSupplyResponse, GetTotalCreditsInPlatformRequest,
    GetTotalCreditsInPlatformResponse, GetVotePollsByEndDateRequest, GetVotePollsByEndDateResponse,
    WaitForStateTransitionResultRequest, WaitForStateTransitionResultResponse,
};
use dapi_grpc::tonic::{Code, Request, Response, Status};
use dpp::version::PlatformVersion;
use std::fmt::Debug;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::thread::sleep;
use std::time::Duration;
use tracing::Instrument;

/// Service to handle platform queries
pub struct QueryService {
    platform: Arc<Platform<DefaultCoreRPC>>,
}

type QueryMethod<RQ, RS> = fn(
    &Platform<DefaultCoreRPC>,
    RQ,
    &PlatformState,
    &PlatformVersion,
) -> Result<QueryValidationResult<RS>, Error>;

impl QueryService {
    /// Creates new QueryService
    pub fn new(platform: Arc<Platform<DefaultCoreRPC>>) -> Self {
        Self { platform }
    }

    async fn handle_blocking_query<RQ, RS>(
        &self,
        request: Request<RQ>,
        query_method: QueryMethod<RQ, RS>,
        endpoint_name: &str,
    ) -> Result<Response<RS>, Status>
    where
        RS: Clone + Send + 'static,
        RQ: Debug + Send + Clone + 'static,
    {
        let mut response_duration_metric = query_duration_metric(endpoint_name);

        let platform = Arc::clone(&self.platform);

        let request_debug = format!("{:?}", &request);

        let result = spawn_blocking_task_with_name_if_supported("query", move || {
            let mut result;

            let query_request = request.into_inner();

            let mut query_counter = 0;

            loop {
                let platform_state = platform.state.load();

                let platform_version = platform_state
                    .current_platform_version()
                    .map_err(|_| Status::unavailable("platform is not initialized"))?;

                // Query is using Platform execution state and Drive state to during the execution.
                // They are updating every block in finalize block ABCI handler.
                // The problem is that these two operations aren't atomic and some latency between
                // them could lead to data races. `committed_block_height_guard` counter that represents
                // the latest the height of latest committed Drive state and logic bellow ensures
                // that query is executed only after/before both states are updated.
                let mut needs_restart = false;

                loop {
                    let committed_block_height_guard = platform
                        .committed_block_height_guard
                        .load(Ordering::Relaxed);
                    let mut counter = 0;
                    if platform_state.last_committed_block_height() == committed_block_height_guard
                    {
                        break;
                    } else {
                        counter += 1;
                        sleep(Duration::from_millis(10))
                    }

                    // We try for up to 1 second
                    if counter >= 100 {
                        query_counter += 1;
                        needs_restart = true;
                        break;
                    }
                }

                if query_counter > 3 {
                    return Err(query_error_into_status(QueryError::NotServiceable(
                        "platform is saturated (did not attempt query)".to_string(),
                    )));
                }

                if needs_restart {
                    continue;
                }

                result = query_method(
                    &platform,
                    query_request.clone(),
                    &platform_state,
                    platform_version,
                );

                let committed_block_height_guard = platform
                    .committed_block_height_guard
                    .load(Ordering::Relaxed);

                if platform_state.last_committed_block_height() == committed_block_height_guard {
                    // in this case the query almost certainly executed correctly
                    break;
                } else {
                    query_counter += 1;

                    if query_counter > 2 {
                        // This should never be possible
                        return Err(query_error_into_status(QueryError::NotServiceable(
                            "platform is saturated".to_string(),
                        )));
                    }
                }
            }

            let mut query_result = result.map_err(error_into_status)?;

            if query_result.is_valid() {
                let response = query_result
                    .into_data()
                    .map_err(|error| error_into_status(error.into()))?;

                Ok(Response::new(response))
            } else {
                let error = query_result.errors.swap_remove(0);

                Err(query_error_into_status(error))
            }
        })?
        .instrument(tracing::trace_span!("query", endpoint_name))
        .await
        .map_err(|error| Status::internal(format!("query thread failed: {}", error)))?;

        // Query logging and metrics
        let code = match &result {
            Ok(_) => Code::Ok,
            Err(status) => status.code(),
        };

        let code_label = format!("{:?}", code).to_lowercase();

        // Add code to response duration metric
        let label = abci_response_code_metric_label(code);
        response_duration_metric.add_label(label);

        match code {
            // User errors
            Code::Ok
            | Code::InvalidArgument
            | Code::NotFound
            | Code::AlreadyExists
            | Code::ResourceExhausted
            | Code::PermissionDenied
            | Code::Unavailable
            | Code::Aborted
            | Code::FailedPrecondition
            | Code::OutOfRange
            | Code::Cancelled
            | Code::DeadlineExceeded
            | Code::Unauthenticated => {
                let elapsed_time = response_duration_metric.elapsed().as_secs_f64();

                tracing::trace!(
                    request = request_debug,
                    elapsed_time,
                    endpoint_name,
                    code = code_label,
                    "query '{}' executed with code {:?} in {} secs",
                    endpoint_name,
                    code,
                    elapsed_time
                );
            }
            // System errors
            Code::Unknown | Code::Unimplemented | Code::Internal | Code::DataLoss => {
                tracing::error!(
                    request = request_debug,
                    endpoint_name,
                    code = code_label,
                    "query '{}' execution failed with code {:?}",
                    endpoint_name,
                    code
                );
            }
        }

        result
    }
}

fn respond_with_unimplemented<RS>(name: &str) -> Result<Response<RS>, Status> {
    tracing::error!("{} endpoint is called but it's not supported", name);

    Err(Status::unimplemented("the endpoint is not supported"))
}

#[async_trait]
impl PlatformService for QueryService {
    async fn broadcast_state_transition(
        &self,
        _request: Request<BroadcastStateTransitionRequest>,
    ) -> Result<Response<BroadcastStateTransitionResponse>, Status> {
        respond_with_unimplemented("broadcast_state_transition")
    }

    async fn get_identity(
        &self,
        request: Request<GetIdentityRequest>,
    ) -> Result<Response<GetIdentityResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_identity,
            "get_identity",
        )
        .await
    }

    async fn get_identities_contract_keys(
        &self,
        request: Request<GetIdentitiesContractKeysRequest>,
    ) -> Result<Response<GetIdentitiesContractKeysResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_identities_contract_keys,
            "get_identities_contract_keys",
        )
        .await
    }

    async fn get_identity_keys(
        &self,
        request: Request<GetIdentityKeysRequest>,
    ) -> Result<Response<GetIdentityKeysResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_keys,
            "get_identity_keys",
        )
        .await
    }

    async fn get_identity_nonce(
        &self,
        request: Request<GetIdentityNonceRequest>,
    ) -> Result<Response<GetIdentityNonceResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_identity_nonce,
            "get_identity_nonce",
        )
        .await
    }

    async fn get_identity_contract_nonce(
        &self,
        request: Request<GetIdentityContractNonceRequest>,
    ) -> Result<Response<GetIdentityContractNonceResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_identity_contract_nonce,
            "get_identity_contract_nonce",
        )
        .await
    }

    async fn get_identity_balance(
        &self,
        request: Request<GetIdentityBalanceRequest>,
    ) -> Result<Response<GetIdentityBalanceResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_balance,
            "get_identity_balance",
        )
        .await
    }

    async fn get_identity_balance_and_revision(
        &self,
        request: Request<GetIdentityBalanceAndRevisionRequest>,
    ) -> Result<Response<GetIdentityBalanceAndRevisionResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_balance_and_revision,
            "get_identity_balance_and_revision",
        )
        .await
    }

    async fn get_data_contract(
        &self,
        request: Request<GetDataContractRequest>,
    ) -> Result<Response<GetDataContractResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_data_contract,
            "get_data_contract",
        )
        .await
    }

    async fn get_data_contract_history(
        &self,
        request: Request<GetDataContractHistoryRequest>,
    ) -> Result<Response<GetDataContractHistoryResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_data_contract_history,
            "get_data_contract_history",
        )
        .await
    }

    async fn get_data_contracts(
        &self,
        request: Request<GetDataContractsRequest>,
    ) -> Result<Response<GetDataContractsResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_data_contracts,
            "get_data_contracts",
        )
        .await
    }

    async fn get_document_history(
        &self,
        request: Request<GetDocumentHistoryRequest>,
    ) -> Result<Response<GetDocumentHistoryResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_document_history,
            "get_document_history",
        )
        .await
    }

    async fn get_documents(
        &self,
        request: Request<GetDocumentsRequest>,
    ) -> Result<Response<GetDocumentsResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_documents,
            "get_documents",
        )
        .await
    }

    async fn get_identity_by_public_key_hash(
        &self,
        request: Request<GetIdentityByPublicKeyHashRequest>,
    ) -> Result<Response<GetIdentityByPublicKeyHashResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_identity_by_public_key_hash,
            "get_identity_by_public_key_hash",
        )
        .await
    }

    async fn get_identity_by_non_unique_public_key_hash(
        &self,
        request: Request<GetIdentityByNonUniquePublicKeyHashRequest>,
    ) -> Result<Response<GetIdentityByNonUniquePublicKeyHashResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_identity_by_non_unique_public_key_hash,
            "get_identity_by_non_unique_public_key_hash",
        )
        .await
    }

    async fn wait_for_state_transition_result(
        &self,
        _request: Request<WaitForStateTransitionResultRequest>,
    ) -> Result<Response<WaitForStateTransitionResultResponse>, Status> {
        respond_with_unimplemented("wait_for_state_transition_result")
    }

    async fn get_consensus_params(
        &self,
        _request: Request<GetConsensusParamsRequest>,
    ) -> Result<Response<GetConsensusParamsResponse>, Status> {
        respond_with_unimplemented("get_consensus_params")
    }

    async fn get_protocol_version_upgrade_state(
        &self,
        request: Request<GetProtocolVersionUpgradeStateRequest>,
    ) -> Result<Response<GetProtocolVersionUpgradeStateResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_version_upgrade_state,
            "get_protocol_version_upgrade_state",
        )
        .await
    }

    async fn get_protocol_version_upgrade_vote_status(
        &self,
        request: Request<GetProtocolVersionUpgradeVoteStatusRequest>,
    ) -> Result<Response<GetProtocolVersionUpgradeVoteStatusResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_version_upgrade_vote_status,
            "get_protocol_version_upgrade_vote_status",
        )
        .await
    }

    async fn get_epochs_info(
        &self,
        request: Request<GetEpochsInfoRequest>,
    ) -> Result<Response<GetEpochsInfoResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_epoch_infos,
            "get_epochs_info",
        )
        .await
    }

    async fn get_path_elements(
        &self,
        request: Request<GetPathElementsRequest>,
    ) -> Result<Response<GetPathElementsResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_path_elements,
            "get_path_elements",
        )
        .await
    }

    async fn get_contested_resources(
        &self,
        request: Request<GetContestedResourcesRequest>,
    ) -> Result<Response<GetContestedResourcesResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_contested_resources,
            "get_contested_resources",
        )
        .await
    }

    async fn get_contested_resource_vote_state(
        &self,
        request: Request<GetContestedResourceVoteStateRequest>,
    ) -> Result<Response<GetContestedResourceVoteStateResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_contested_resource_vote_state,
            "get_contested_resource_vote_state",
        )
        .await
    }

    async fn get_contested_resource_voters_for_identity(
        &self,
        request: Request<GetContestedResourceVotersForIdentityRequest>,
    ) -> Result<Response<GetContestedResourceVotersForIdentityResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_contested_resource_voters_for_identity,
            "get_contested_resource_voters_for_identity",
        )
        .await
    }

    async fn get_contested_resource_identity_votes(
        &self,
        request: Request<GetContestedResourceIdentityVotesRequest>,
    ) -> Result<Response<GetContestedResourceIdentityVotesResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_contested_resource_identity_votes,
            "get_contested_resource_identity_votes",
        )
        .await
    }

    async fn get_vote_polls_by_end_date(
        &self,
        request: Request<GetVotePollsByEndDateRequest>,
    ) -> Result<Response<GetVotePollsByEndDateResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_vote_polls_by_end_date_query,
            "get_vote_polls_by_end_date",
        )
        .await
    }

    async fn get_prefunded_specialized_balance(
        &self,
        request: Request<GetPrefundedSpecializedBalanceRequest>,
    ) -> Result<Response<GetPrefundedSpecializedBalanceResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_prefunded_specialized_balance,
            "get_prefunded_specialized_balance",
        )
        .await
    }

    async fn get_total_credits_in_platform(
        &self,
        request: Request<GetTotalCreditsInPlatformRequest>,
    ) -> Result<Response<GetTotalCreditsInPlatformResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_total_credits_in_platform,
            "get_total_credits_in_platform",
        )
        .await
    }

    async fn get_identities_balances(
        &self,
        request: Request<GetIdentitiesBalancesRequest>,
    ) -> Result<Response<GetIdentitiesBalancesResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_identities_balances,
            "get_identities_balances",
        )
        .await
    }

    async fn get_status(
        &self,
        request: Request<GetStatusRequest>,
    ) -> Result<Response<GetStatusResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_partial_status,
            "query_partial_status",
        )
        .await
    }

    async fn get_evonodes_proposed_epoch_blocks_by_ids(
        &self,
        request: Request<GetEvonodesProposedEpochBlocksByIdsRequest>,
    ) -> Result<Response<GetEvonodesProposedEpochBlocksResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_proposed_block_counts_by_evonode_ids,
            "query_proposed_block_counts_by_evonode_ids",
        )
        .await
    }

    async fn get_evonodes_proposed_epoch_blocks_by_range(
        &self,
        request: Request<GetEvonodesProposedEpochBlocksByRangeRequest>,
    ) -> Result<Response<GetEvonodesProposedEpochBlocksResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_proposed_block_counts_by_range,
            "query_proposed_block_counts_by_range",
        )
        .await
    }

    async fn get_current_quorums_info(
        &self,
        request: Request<GetCurrentQuorumsInfoRequest>,
    ) -> Result<Response<GetCurrentQuorumsInfoResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_current_quorums_info,
            "query_current_quorums_info",
        )
        .await
    }

    async fn get_identity_token_balances(
        &self,
        request: Request<GetIdentityTokenBalancesRequest>,
    ) -> Result<Response<GetIdentityTokenBalancesResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_identity_token_balances,
            "query_identity_token_balances",
        )
        .await
    }

    async fn get_identities_token_balances(
        &self,
        request: Request<GetIdentitiesTokenBalancesRequest>,
    ) -> Result<Response<GetIdentitiesTokenBalancesResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_identities_token_balances,
            "query_identities_token_balances",
        )
        .await
    }

    async fn get_identity_token_infos(
        &self,
        request: Request<GetIdentityTokenInfosRequest>,
    ) -> Result<Response<GetIdentityTokenInfosResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_identity_token_infos,
            "query_identity_token_infos",
        )
        .await
    }

    async fn get_identities_token_infos(
        &self,
        request: Request<GetIdentitiesTokenInfosRequest>,
    ) -> Result<Response<GetIdentitiesTokenInfosResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_identities_token_infos,
            "query_identities_token_infos",
        )
        .await
    }

    async fn get_token_statuses(
        &self,
        request: Request<GetTokenStatusesRequest>,
    ) -> Result<Response<GetTokenStatusesResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_token_statuses,
            "get_token_statuses",
        )
        .await
    }

    async fn get_token_pre_programmed_distributions(
        &self,
        request: Request<GetTokenPreProgrammedDistributionsRequest>,
    ) -> Result<Response<GetTokenPreProgrammedDistributionsResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_token_pre_programmed_distributions,
            "get_token_pre_programmed_distributions",
        )
        .await
    }

    async fn get_token_total_supply(
        &self,
        request: Request<GetTokenTotalSupplyRequest>,
    ) -> Result<Response<GetTokenTotalSupplyResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_token_total_supply,
            "get_token_total_supply",
        )
        .await
    }

    async fn get_group_info(
        &self,
        request: Request<GetGroupInfoRequest>,
    ) -> Result<Response<GetGroupInfoResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_group_info,
            "get_group_info",
        )
        .await
    }

    async fn get_group_infos(
        &self,
        request: Request<GetGroupInfosRequest>,
    ) -> Result<Response<GetGroupInfosResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_group_infos,
            "get_group_infos",
        )
        .await
    }

    async fn get_group_actions(
        &self,
        request: Request<GetGroupActionsRequest>,
    ) -> Result<Response<GetGroupActionsResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_group_actions,
            "get_group_actions",
        )
        .await
    }

    async fn get_group_action_signers(
        &self,
        request: Request<GetGroupActionSignersRequest>,
    ) -> Result<Response<GetGroupActionSignersResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_group_action_signers,
            "get_group_action_signers",
        )
        .await
    }

    async fn get_token_direct_purchase_prices(
        &self,
        request: Request<GetTokenDirectPurchasePricesRequest>,
    ) -> Result<Response<GetTokenDirectPurchasePricesResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_token_direct_purchase_prices,
            "get_token_direct_purchase_prices",
        )
        .await
    }

    async fn get_token_contract_info(
        &self,
        request: Request<GetTokenContractInfoRequest>,
    ) -> Result<Response<GetTokenContractInfoResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_token_contract_info,
            "get_token_contract_info",
        )
        .await
    }

    async fn get_token_perpetual_distribution_last_claim(
        &self,
        request: Request<GetTokenPerpetualDistributionLastClaimRequest>,
    ) -> Result<Response<GetTokenPerpetualDistributionLastClaimResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_token_perpetual_distribution_last_claim,
            "get_token_perpetual_distribution_last_claim",
        )
        .await
    }

    async fn get_finalized_epoch_infos(
        &self,
        request: Request<GetFinalizedEpochInfosRequest>,
    ) -> Result<Response<GetFinalizedEpochInfosResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_finalized_epoch_infos,
            "get_finalized_epoch_infos",
        )
        .await
    }

    async fn get_address_info(
        &self,
        request: Request<GetAddressInfoRequest>,
    ) -> Result<Response<GetAddressInfoResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_address_info,
            "get_address_info",
        )
        .await
    }

    async fn get_addresses_infos(
        &self,
        request: Request<GetAddressesInfosRequest>,
    ) -> Result<Response<GetAddressesInfosResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_addresses_infos,
            "get_addresses_infos",
        )
        .await
    }

    async fn get_addresses_trunk_state(
        &self,
        request: Request<GetAddressesTrunkStateRequest>,
    ) -> Result<Response<GetAddressesTrunkStateResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_addresses_trunk_state,
            "get_addresses_trunk_state",
        )
        .await
    }

    async fn get_addresses_branch_state(
        &self,
        request: Request<GetAddressesBranchStateRequest>,
    ) -> Result<Response<GetAddressesBranchStateResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_addresses_branch_state,
            "get_addresses_branch_state",
        )
        .await
    }

    async fn get_recent_address_balance_changes(
        &self,
        request: Request<GetRecentAddressBalanceChangesRequest>,
    ) -> Result<Response<GetRecentAddressBalanceChangesResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_recent_address_balance_changes,
            "get_recent_address_balance_changes",
        )
        .await
    }

    async fn get_recent_compacted_address_balance_changes(
        &self,
        request: Request<GetRecentCompactedAddressBalanceChangesRequest>,
    ) -> Result<Response<GetRecentCompactedAddressBalanceChangesResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_recent_compacted_address_balance_changes,
            "get_recent_compacted_address_balance_changes",
        )
        .await
    }

    async fn get_shielded_encrypted_notes(
        &self,
        request: Request<GetShieldedEncryptedNotesRequest>,
    ) -> Result<Response<GetShieldedEncryptedNotesResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_shielded_encrypted_notes,
            "get_shielded_encrypted_notes",
        )
        .await
    }

    async fn get_shielded_anchors(
        &self,
        request: Request<GetShieldedAnchorsRequest>,
    ) -> Result<Response<GetShieldedAnchorsResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_shielded_anchors,
            "get_shielded_anchors",
        )
        .await
    }

    async fn get_most_recent_shielded_anchor(
        &self,
        request: Request<GetMostRecentShieldedAnchorRequest>,
    ) -> Result<Response<GetMostRecentShieldedAnchorResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_most_recent_shielded_anchor,
            "get_most_recent_shielded_anchor",
        )
        .await
    }

    async fn get_shielded_pool_state(
        &self,
        request: Request<GetShieldedPoolStateRequest>,
    ) -> Result<Response<GetShieldedPoolStateResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_shielded_pool_state,
            "get_shielded_pool_state",
        )
        .await
    }

    async fn get_shielded_notes_count(
        &self,
        request: Request<GetShieldedNotesCountRequest>,
    ) -> Result<Response<GetShieldedNotesCountResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_shielded_notes_count,
            "get_shielded_notes_count",
        )
        .await
    }

    async fn get_shielded_nullifiers(
        &self,
        request: Request<GetShieldedNullifiersRequest>,
    ) -> Result<Response<GetShieldedNullifiersResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_shielded_nullifiers,
            "get_shielded_nullifiers",
        )
        .await
    }

    async fn get_nullifiers_trunk_state(
        &self,
        request: Request<GetNullifiersTrunkStateRequest>,
    ) -> Result<Response<GetNullifiersTrunkStateResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_nullifiers_trunk_state,
            "get_nullifiers_trunk_state",
        )
        .await
    }

    async fn get_nullifiers_branch_state(
        &self,
        request: Request<GetNullifiersBranchStateRequest>,
    ) -> Result<Response<GetNullifiersBranchStateResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_nullifiers_branch_state,
            "get_nullifiers_branch_state",
        )
        .await
    }

    async fn get_recent_nullifier_changes(
        &self,
        request: Request<GetRecentNullifierChangesRequest>,
    ) -> Result<Response<GetRecentNullifierChangesResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_recent_nullifier_changes,
            "get_recent_nullifier_changes",
        )
        .await
    }

    async fn get_recent_compacted_nullifier_changes(
        &self,
        request: Request<GetRecentCompactedNullifierChangesRequest>,
    ) -> Result<Response<GetRecentCompactedNullifierChangesResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_recent_compacted_nullifier_changes,
            "get_recent_compacted_nullifier_changes",
        )
        .await
    }
}

#[async_trait]
impl DriveInternal for QueryService {
    async fn get_proofs(
        &self,
        request: Request<GetProofsRequest>,
    ) -> Result<Response<GetProofsResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_proofs,
            "get_proofs",
        )
        .await
    }
}

fn query_error_into_status(error: QueryError) -> Status {
    match error {
        QueryError::NotFound(message) => Status::not_found(message),
        QueryError::InvalidArgument(message) => Status::invalid_argument(message),
        QueryError::Query(error) => Status::invalid_argument(error.to_string()),
        _ => {
            tracing::error!("unexpected query error: {:?}", error);

            Status::unknown(error.to_string())
        }
    }
}

fn error_into_status(error: Error) -> Status {
    Status::internal(format!("query: {}", error))
}