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,
    GetContractDocumentRemovalsRequest, GetContractDocumentRemovalsResponse,
    GetContractFeePotsRequest, GetContractFeePotsResponse, GetContractGroupInfoRequest,
    GetContractGroupInfoResponse, GetContractGroupMembersRequest, GetContractGroupMembersResponse,
    GetContractGroupsForContractRequest, GetContractGroupsForContractResponse,
    GetContractModerationActionCountsRequest, GetContractModerationActionCountsResponse,
    GetContractModerationEntriesRequest, GetContractModerationEntriesResponse,
    GetContractModerationStatusRequest, GetContractModerationStatusResponse,
    GetContractTeamActionSignersRequest, GetContractTeamActionSignersResponse,
    GetContractTeamActionsRequest, GetContractTeamActionsResponse, GetCurrentQuorumsInfoRequest,
    GetCurrentQuorumsInfoResponse, GetDataContractHistoryRequest, GetDataContractHistoryResponse,
    GetDataContractRequest, GetDataContractResponse, GetDataContractsByRangeRequest,
    GetDataContractsLatestVersionsRequest, GetDataContractsLatestVersionsResponse,
    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,
    GetIdentityKeysRemainingBudgetsRequest, GetIdentityKeysRemainingBudgetsResponse,
    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 drive::error::Error as DriveError;
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;
/// Longest part of a status message the service keeps; a longer message is cut there and gets
/// a short note with the number of bytes dropped. Tonic sends the message in the
/// `grpc-message` header, and an HTTP/2 peer refuses an oversized header list, which its client
/// then reports as an internal error. A message can echo a request field of any length, such
/// as a document type name.
const MAX_STATUS_MESSAGE_BYTES: usize = 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(),
                        )));
                    }
                }
            }

            query_result_into_response(result).map(Response::new)
        })?
        .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_keys_remaining_budgets(
        &self,
        request: Request<GetIdentityKeysRemainingBudgetsRequest>,
    ) -> Result<Response<GetIdentityKeysRemainingBudgetsResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_identity_keys_remaining_budgets,
            "get_identity_keys_remaining_budgets",
        )
        .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_data_contracts_by_range(
        &self,
        request: Request<GetDataContractsByRangeRequest>,
    ) -> Result<Response<GetDataContractsResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_data_contracts_by_range,
            "get_data_contracts_by_range",
        )
        .await
    }

    async fn get_data_contracts_latest_versions(
        &self,
        request: Request<GetDataContractsLatestVersionsRequest>,
    ) -> Result<Response<GetDataContractsLatestVersionsResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_data_contracts_latest_versions,
            "get_data_contracts_latest_versions",
        )
        .await
    }

    async fn get_contract_group_info(
        &self,
        request: Request<GetContractGroupInfoRequest>,
    ) -> Result<Response<GetContractGroupInfoResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_contract_group_info,
            "get_contract_group_info",
        )
        .await
    }

    async fn get_contract_moderation_status(
        &self,
        request: Request<GetContractModerationStatusRequest>,
    ) -> Result<Response<GetContractModerationStatusResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_contract_moderation_status,
            "get_contract_moderation_status",
        )
        .await
    }

    async fn get_contract_document_removals(
        &self,
        request: Request<GetContractDocumentRemovalsRequest>,
    ) -> Result<Response<GetContractDocumentRemovalsResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_contract_document_removals,
            "get_contract_document_removals",
        )
        .await
    }

    async fn get_contract_team_actions(
        &self,
        request: Request<GetContractTeamActionsRequest>,
    ) -> Result<Response<GetContractTeamActionsResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_contract_team_actions,
            "get_contract_team_actions",
        )
        .await
    }

    async fn get_contract_team_action_signers(
        &self,
        request: Request<GetContractTeamActionSignersRequest>,
    ) -> Result<Response<GetContractTeamActionSignersResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_contract_team_action_signers,
            "get_contract_team_action_signers",
        )
        .await
    }

    async fn get_contract_moderation_action_counts(
        &self,
        request: Request<GetContractModerationActionCountsRequest>,
    ) -> Result<Response<GetContractModerationActionCountsResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_contract_moderation_action_counts,
            "get_contract_moderation_action_counts",
        )
        .await
    }

    async fn get_contract_moderation_entries(
        &self,
        request: Request<GetContractModerationEntriesRequest>,
    ) -> Result<Response<GetContractModerationEntriesResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_contract_moderation_entries,
            "get_contract_moderation_entries",
        )
        .await
    }

    async fn get_contract_fee_pots(
        &self,
        request: Request<GetContractFeePotsRequest>,
    ) -> Result<Response<GetContractFeePotsResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_contract_fee_pots,
            "get_contract_fee_pots",
        )
        .await
    }

    async fn get_contract_group_members(
        &self,
        request: Request<GetContractGroupMembersRequest>,
    ) -> Result<Response<GetContractGroupMembersResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_contract_group_members,
            "get_contract_group_members",
        )
        .await
    }

    async fn get_contract_groups_for_contract(
        &self,
        request: Request<GetContractGroupsForContractRequest>,
    ) -> Result<Response<GetContractGroupsForContractResponse>, Status> {
        self.handle_blocking_query(
            request,
            Platform::<DefaultCoreRPC>::query_contract_groups_for_contract,
            "get_contract_groups_for_contract",
        )
        .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(bounded_message(message)),
        QueryError::InvalidArgument(message) => Status::invalid_argument(bounded_message(message)),
        QueryError::Query(error) | QueryError::Drive(DriveError::Query(error)) => {
            Status::invalid_argument(bounded_message(error.to_string()))
        }
        QueryError::TooManyElements(message) => Status::invalid_argument(bounded_message(message)),
        QueryError::ResourceExhausted(message) => {
            Status::resource_exhausted(bounded_message(message))
        }
        // A request this node cannot decode, or a request version it does not serve, may be
        // one a newer node serves, so these stay retryable on another node.
        _ => {
            tracing::error!("unexpected query error: {:?}", error);

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

/// The response, or the status, the service answers for a query method's result.
pub(super) fn query_result_into_response<T: Clone>(
    result: Result<QueryValidationResult<T>, Error>,
) -> Result<T, Status> {
    let mut query_result = result.map_err(error_into_status)?;

    if query_result.is_valid() {
        query_result
            .into_data()
            .map_err(|error| error_into_status(error.into()))
    } else {
        let error = query_result.errors.swap_remove(0);

        Err(query_error_into_status(error))
    }
}

fn error_into_status(error: Error) -> Status {
    match error {
        // A query syntax error is the request's fault whichever layer finds it. As an internal
        // error, clients would retry it on every other node and stop using each one.
        Error::Drive(DriveError::Query(error)) => {
            Status::invalid_argument(bounded_message(error.to_string()))
        }
        error => Status::internal(bounded_message(format!("query: {}", error))),
    }
}

fn bounded_message(mut message: String) -> String {
    if message.len() <= MAX_STATUS_MESSAGE_BYTES {
        return message;
    }
    let mut end = MAX_STATUS_MESSAGE_BYTES;
    while !message.is_char_boundary(end) {
        end -= 1;
    }
    let truncated = message.len() - end;
    message.truncate(end);
    message.push_str(&format!("... ({} bytes truncated)", truncated));
    message
}

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::*;
    use crate::query::tests::{assert_invalid_argument_status, setup_platform};
    use dapi_grpc::platform::v0::get_documents_request;
    use dapi_grpc::platform::v0::get_documents_request::GetDocumentsRequestV0;
    use dpp::dashcore::Network;
    use dpp::system_data_contracts::dpns_contract;
    use drive::error::drive::DriveError as DriveInternalError;
    use drive::error::query::QuerySyntaxError;

    #[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);
    }

    /// A query syntax error is the request's fault whether the handler or Drive found it.
    #[test]
    fn should_answer_a_query_syntax_error_from_drive_with_invalid_argument() {
        let syntax_error = || QuerySyntaxError::InvalidLimit("limit 0".to_string());

        let from_result = error_into_status(Error::Drive(DriveError::Query(syntax_error())));
        assert_eq!(from_result.code(), Code::InvalidArgument);
        assert!(from_result.message().contains("limit 0"));

        let from_validation =
            query_error_into_status(QueryError::Drive(DriveError::Query(syntax_error())));
        assert_eq!(from_validation.code(), Code::InvalidArgument);
    }

    /// What the request did not cause stays retryable on another node: a fault inside the node,
    /// or a request this node cannot decode, which a newer node may serve.
    #[test]
    fn should_keep_node_faults_and_undecodable_requests_retryable() {
        let internal = error_into_status(Error::Drive(DriveError::Drive(
            DriveInternalError::CorruptedCodeExecution("broken"),
        )));
        assert_eq!(internal.code(), Code::Internal);

        let undecodable = query_error_into_status(QueryError::DecodingError(
            "could not decode data contracts query".to_string(),
        ));
        assert_eq!(undecodable.code(), Code::Unknown);
    }

    #[test]
    fn should_bound_a_long_status_message_at_a_char_boundary() {
        // A two-byte character straddles the bound, so the cut falls one byte short of it.
        let message = format!(
            "{}{}",
            "a".repeat(MAX_STATUS_MESSAGE_BYTES - 1),
            "é".repeat(100)
        );

        let status = query_error_into_status(QueryError::InvalidArgument(message.clone()));

        assert_eq!(status.code(), Code::InvalidArgument);
        let kept = &message[..MAX_STATUS_MESSAGE_BYTES - 1];
        assert_eq!(
            status.message(),
            format!(
                "{}... ({} bytes truncated)",
                kept,
                message.len() - kept.len()
            )
        );

        let short = query_error_into_status(QueryError::NotFound("absent".to_string()));
        assert_eq!(short.message(), "absent");
    }

    /// An error echoing a 100 KB document type name used to go out whole in the
    /// `grpc-message` header, past what HTTP/2 peers accept.
    #[test]
    fn should_bound_an_error_that_echoes_an_oversized_document_type_name() {
        let (platform, state, version) = setup_platform(Some((1, 1)), Network::Testnet, None);

        let request = GetDocumentsRequest {
            version: Some(get_documents_request::Version::V0(GetDocumentsRequestV0 {
                data_contract_id: dpns_contract::ID.to_vec(),
                document_type: "x".repeat(100_000),
                r#where: vec![],
                order_by: vec![],
                limit: 0,
                start: None,
                prove: false,
            })),
        };

        let status =
            assert_invalid_argument_status(platform.query_documents(request, &state, version));
        assert!(
            status.message().len() < MAX_STATUS_MESSAGE_BYTES + 64,
            "message is {} bytes",
            status.message().len()
        );
    }
}