I have a Python app - it is a Worker which is read-write part of my CQRS application. It has method that should apply changes simultaniously to a few DB tables + Redis cache. For DB I wrap everything in transaction and inside of this transaction I also update Redis cache. In case if something went wrong on DB side it will roll back all the changes for all the Tables, it also rollback DB changes if anything goes wrong with Redis update, but due to nature of transaction wrappers which are basically a Unit of work, if evrything is updated in Redis already the DB changes are applied only after it, so it means there is one case where the data can become inconsistent because Redis will be updated but DB isn't. Is there any way to maybe make a rollback for Redis in that case? How usually those cases resolved in highload applications? Here is my processor's code that applies those changes:
from typing import Optional

from diamonds_domain.constants import ActionType
from diamonds_domain.database import database
from diamonds_domain.event.repositories import ClaimEventRepository
from diamonds_domain.exceptions import ClaimEventIdempotencyError
from diamonds_domain.profile.repositories import ActionRepository, ProfileRepository
from diamonds_domain.pulsar_manager import (
    PulsarClaimEvent,
    PulsarCommandEvent,
    PulsarCommandType,
    PulsarCreateProfileEvent,
    PulsarManager,
    PulsarUpdateEvent,
)
from diamonds_domain.redis_manager import RedisManager


class EventProcessor:
    def __init__(
        self,
        pulsar: PulsarManager,
        redis: RedisManager,
        profile_repository: ProfileRepository,
        action_repository: ActionRepository,
        event_repository: ClaimEventRepository,
    ):
        self._redis_manager = redis
        self._pulsar_manager = pulsar
        self._action_repository = action_repository
        self._event_repository = event_repository
        self._profile_repository = profile_repository

    async def process(self, event: PulsarCommandEvent):
        if event.payload.command == PulsarCommandType.CLAIM:
            update = await self._process_claim(event.payload)
        else:
            update = await self._process_profile_create(event.payload)

        if update:
            await self._produce_update(update)

    async def _process_claim(
        self, event: PulsarClaimEvent
    ) -> Optional[PulsarUpdateEvent]:
        profile = await self._profile_repository.get_by_auth_user_uuid(
            event.auth_user_uuid
        )
        if profile is None:
            return None

        try:
            async with database.transaction():
                await self._event_repository.create(
                    auth_user_uuid=event.auth_user_uuid,
                    action=event.action,
                    idempotency_key=event.idempotency_key,
                )

                if event.action == ActionType.CRYSTAL_BALL_ACQUIRE:
                    profile.crystal_balls = profile.crystal_balls + event.value
                elif event.action == ActionType.REF_POINTS_ACQUIRE:
                    profile.ref_points = profile.ref_points + event.value
                elif event.action == ActionType.REF_POINTS_CLAIM:
                    profile.ref_points = profile.ref_points - event.value
                    profile.direct_points = profile.direct_points + event.value
                elif event.action == ActionType.CRYSTAL_BALL_CLAIM:
                    profile.direct_points = profile.direct_points + event.value
                    profile.crystal_balls = profile.crystal_balls - 1
                elif event.action == ActionType.REFERRAL_ACQUIRE:
                    profile.ref_quantity = profile.ref_quantity + 1
                else:
                    profile.direct_points = profile.direct_points + event.value

                await self._profile_repository.update_claimed_values(profile)
                await self._action_repository.update_idempotency_key(
                    action=event.action, profile_uuid=profile.uuid
                )
                actions = await self._action_repository.get_actions_by_profile_uuid(
                    profile.uuid
                )

                await self._redis_manager.set_cached_profile(profile, actions)
                streak = await self._redis_manager.get_streak(
                    auth_user_uuid=profile.auth_user_uuid
                )
        except ClaimEventIdempotencyError:
            return None

        return PulsarUpdateEvent(
            profile_uuid=profile.uuid,
            auth_user_uuid=profile.auth_user_uuid,
            ref_code=profile.ref_code,
            ref_quantity=profile.ref_quantity,
            name=profile.name,
            direct_points=profile.direct_points,
            ref_points=profile.ref_points,
            crystal_balls=profile.crystal_balls,
            streak=streak,
            actions={item.action: item.updated_at for item in actions},
        )

    async def _process_profile_create(
        self, event: PulsarCreateProfileEvent
    ) -> PulsarUpdateEvent:
        profile = await self._profile_repository.create(
            auth_user_uuid=event.auth_user_uuid,
            ref_code=event.ref_code,
            name=event.name,
            direct_referrer_uuid=event.direct_referrer_uuid,
            indirect_referrer_uuid=event.indirect_referrer_uuid,
        )
        actions = await self._action_repository.create_initial_actions(profile.uuid)
        await self._redis_manager.set_cached_profile(profile, actions)
        streak = await self._redis_manager.get_streak(
            auth_user_uuid=profile.auth_user_uuid
        )

        return PulsarUpdateEvent(
            profile_uuid=profile.uuid,
            auth_user_uuid=profile.auth_user_uuid,
            ref_code=profile.ref_code,
            ref_quantity=profile.ref_quantity,
            name=profile.name,
            direct_points=profile.direct_points,
            ref_points=profile.ref_points,
            crystal_balls=profile.crystal_balls,
            streak=streak,
            actions={item.action: item.updated_at for item in actions},
        )

    async def _produce_update(self, update: PulsarUpdateEvent):
        self._pulsar_manager.produce_update(update)