def METHOD_NAME(
    self, metric_path_key: UseCaseKey, messages: Mapping[PartitionIdxOffset, InboundMessage]
) -> CardinalityLimiterState:
    request_hashes = defaultdict(set)
    hash_to_offset: Mapping[str, Dict[int, PartitionIdxOffset]] = defaultdict(dict)
    prefix_to_quota = {}
    # this works by applying one cardinality limiter rollout option
    # for each metric path. ultimately, this can be moved into the
    # loop below to make rollout options occur on a per use case-basis
    rollout_option = {
        UseCaseKey.PERFORMANCE: "sentry-metrics.cardinality-limiter.orgs-rollout-rate",
        UseCaseKey.RELEASE_HEALTH: "sentry-metrics.cardinality-limiter-rh.orgs-rollout-rate",
    }[metric_path_key]
    for key, message in messages.items():
        org_id = message["org_id"]
        if not sample_modulo(rollout_option, org_id):
            continue
        message_hash = int(
            hash_values(
                [
                    message["name"],
                    message["tags"],
                ]
            ),
            16,
        )
        prefix = _build_quota_key(message["use_case_id"], org_id)
        hash_to_offset[prefix][message_hash] = key
        request_hashes[prefix].add(message_hash)
        configured_quota = _construct_quotas(message["use_case_id"])
        # since we might have some use cases that are covered by
        # a quota and some that are not, only add entries that
        # are covered by a quota
        if configured_quota is not None:
            prefix_to_quota[prefix] = configured_quota
    requested_quotas = []
    grants = None
    timestamp = None
    # if none of the use cases are covered by a quota
    if len(prefix_to_quota) == 0:
        return CardinalityLimiterState(
            _cardinality_limiter=self.backend,
            _metric_path_key=metric_path_key,
            _grants=grants,
            _timestamp=timestamp,
            keys_to_remove=[],
        )
    for prefix, hashes in request_hashes.items():
        quota = prefix_to_quota[prefix]
        requested_quotas.append(RequestedQuota(prefix=prefix, unit_hashes=hashes, quota=quota))
    timestamp, grants = self.backend.check_within_quotas(requested_quotas)
    keys_to_remove = hash_to_offset
    # make sure that hash_to_offset is no longer used, as the underlying
    # dict will be mutated
    del hash_to_offset
    for grant in grants:
        for hash in grant.granted_unit_hashes:
            del keys_to_remove[grant.request.prefix][hash]
        substrings = grant.request.prefix.split("-")
        grant_use_case_id = substrings[3]
        grant_org_id = substrings[-1]
        metrics.incr(
            "sentry_metrics.indexer.process_messages.dropped_message",
            amount=len(keys_to_remove[grant.request.prefix]),
            tags={
                "reason": "cardinality_limit",
                "use_case_id": grant_use_case_id,
                "org_id": grant_org_id,
            },
        )
    return CardinalityLimiterState(
        _cardinality_limiter=self.backend,
        _metric_path_key=metric_path_key,
        _grants=grants,
        _timestamp=timestamp,
        keys_to_remove=[key for grant in keys_to_remove.values() for key in grant.values()],
    )