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::get_path_elements_request;
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,
    GetPathElementsRequest, GetPathElementsResponse, GetPrefundedSpecializedBalanceRequest,
    GetPrefundedSpecializedBalanceResponse, GetProtocolVersionUpgradeStateRequest,
    GetProtocolVersionUpgradeStateResponse, GetProtocolVersionUpgradeVoteStatusRequest,
    GetProtocolVersionUpgradeVoteStatusResponse, GetRecentAddressBalanceChangesRequest,
    GetRecentAddressBalanceChangesResponse, GetRecentCompactedAddressBalanceChangesRequest,
    GetRecentCompactedAddressBalanceChangesResponse, 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::sync::atomic::Ordering;
use std::sync::Arc;
use std::thread::sleep;
use std::time::Duration;
use tracing::Instrument;

const MAX_PATH_COMPONENTS: usize = 256;
const MAX_GROVEDB_KEY_BYTES: usize = 255;
const MAX_PATH_QUERY_BYTES: usize = 64 * 1024;

/// 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: Send + Clone + 'static,
    {
        let mut response_duration_metric = query_duration_metric(endpoint_name);

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

        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!(
                    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!(
                    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> {
        validate_path_elements_request(request.get_ref())?;
        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_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()),
        QueryError::ResourceExhausted(message) => Status::resource_exhausted(message),
        _ => {
            tracing::error!("unexpected query error: {:?}", error);

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

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

fn validate_path_elements_request(request: &GetPathElementsRequest) -> Result<(), Status> {
    let v0 = match request.version.as_ref() {
        Some(get_path_elements_request::Version::V0(v0)) => v0,
        None => return Err(Status::invalid_argument("missing request version")),
    };
    let max_keys = PlatformVersion::latest()
        .drive_abci
        .query
        .max_returned_elements as usize;
    if v0.path.len() > MAX_PATH_COMPONENTS || v0.keys.len() > max_keys {
        return Err(Status::resource_exhausted(
            "too many path or key components",
        ));
    }

    let total_bytes = v0
        .path
        .iter()
        .chain(&v0.keys)
        .try_fold(0usize, |total, component| {
            if component.len() > MAX_GROVEDB_KEY_BYTES {
                return Err(Status::resource_exhausted(
                    "path or key component is too large",
                ));
            }
            total
                .checked_add(component.len())
                .ok_or_else(|| Status::resource_exhausted("path query size overflow"))
        })?;

    if total_bytes > MAX_PATH_QUERY_BYTES {
        return Err(Status::resource_exhausted(
            "aggregate path query bytes exceed limit",
        ));
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use dapi_grpc::platform::v0::get_path_elements_request::GetPathElementsRequestV0;

    #[test]
    fn path_elements_request_is_bounded_before_debug_formatting() {
        let request = GetPathElementsRequest {
            version: Some(get_path_elements_request::Version::V0(
                GetPathElementsRequestV0 {
                    path: vec![vec![]; MAX_PATH_COMPONENTS + 1],
                    keys: vec![],
                    prove: false,
                },
            )),
        };

        let status = validate_path_elements_request(&request)
            .expect_err("expected excessive path depth to be rejected");
        assert_eq!(status.code(), Code::ResourceExhausted);
    }
}

#[cfg(test)]
mod query_error_status_tests {
    use super::*;

    #[test]
    fn resource_exhausted_query_error_maps_to_retryable_grpc_status() {
        let status = query_error_into_status(QueryError::ResourceExhausted(
            "server-side retained state is over capacity".to_string(),
        ));

        assert_eq!(status.code(), Code::ResourceExhausted);
    }
}