diff --git a/app/api/task_utils.py b/app/api/task_utils.py new file mode 100644 index 0000000..ea8a1a6 --- /dev/null +++ b/app/api/task_utils.py @@ -0,0 +1,86 @@ +from loguru import logger +from redis.asyncio import Redis + +from app.api.error_report.report import Frame, Report +from app.api.utils import redis_delete, redis_get_raw, redis_set +from app.external.result_type.src.result.result import Err, Ok, Result + + +async def get_lock_status(key: str, redis: Redis | None = None) -> Result[bool, Report]: + """ + Checks if the update lock is held. + Returns True if lock is held. + """ + logger.trace(f"get_lock_status({key})") + + result = await redis_get_raw(key, custom_redis=redis) + match result: + case Ok(None): + logger.debug(f"Lock for key {key} not held.") + return Ok(False) + case Ok(data) if isinstance(data, bytes) and data.decode("utf-8") == "locked": + logger.debug(f"Lock held for key {key}.") + return Ok(True) + case Ok(data) if not isinstance(data, bytes): + return Err(Report(message=f"Unexpected data from Redis: {data}")) + case Err(e): + logger.error(f"Error checking if lock is held for key {key}: %s", e) + return Err(e) + + return Err(Report(message=f"Error checking if lock is held for key {key}")) + + +async def acquire_lock( + key: str, lock_ttl: int, redis: Redis | None = None +) -> Result[bool, Report]: + """ + Attempts to acquire a lock to prevent concurrent updates. + Returns True if lock acquired. + """ + logger.trace(f"acquire_lock({key})") + + result = await get_lock_status(key, redis) + match result: + case Ok(lock_status): + if lock_status: + return Ok(False) + case Err(report): + return Err(report) + + # nx: Only set the key if it does not already exist. + # ex: Set the specified expire time, in seconds. + # TODO: should this even timeout automatically? + # do we need to do some cleanup if the status update takes too long? + match await redis_set( + key, + "locked", + nx=True, + ex=lock_ttl, + custom_redis=redis, + ): + case Ok(_): + return Ok(True) + case Err(report): + return Err( + report.attach_frame( + Frame(message="Error acquiring update lock in Redis") + ) + ) + + return Err(Report(message="Error acquiring update lock in Redis")) + + +async def release_update_lock( + key: str, redis: Redis | None = None +) -> Result[None, Report]: + """Releases the update lock.""" + logger.trace(f"release_update_lock({key})") + match await redis_delete(key, custom_redis=redis): + case Ok(_): + return Ok(None) + case Err(report): + return Err( + report.attach_frame( + Frame(message=f"Error releasing lock in Redis for {key}") + ) + ) diff --git a/app/api/warmup.py b/app/api/warmup.py index 618316e..ee8c223 100644 --- a/app/api/warmup.py +++ b/app/api/warmup.py @@ -5,7 +5,9 @@ from fastapi import HTTPException, status from loguru import logger from app.api.error_report.report import Report -from app.apps.cache import get_cached_app_status, get_lock_status +from app.api.task_utils import get_lock_status +from app.apps.cache import get_cached_app_status +from app.apps.constants import AppsServiceKeys from app.apps.models import AppStatusQueryResult from app.apps.tasks import update_app_state_task from app.bitcoind.service import get_btc_info @@ -43,7 +45,7 @@ async def _get_app_status_data() -> Result[Optional[AppStatusQueryResult], Repor # TODO: return error message logger.error(f"Failed to fetch app status: {report.format_verbose()}") - result = await get_lock_status() + result = await get_lock_status(AppsServiceKeys.APP_STATUS_LOCK_KEY) match result: case Ok(True): logger.info( diff --git a/app/apps/cache.py b/app/apps/cache.py index ca9a756..312efba 100644 --- a/app/apps/cache.py +++ b/app/apps/cache.py @@ -6,16 +6,9 @@ from redis.asyncio import Redis from app.api.channel import BaseChannelListener from app.api.config import config -from app.api.error_report.report import Frame, Report +from app.api.error_report.report import Report from app.api.models import ErrorMessage -from app.api.utils import ( - SSE, - broadcast_sse_msg, - redis_delete, - redis_get, - redis_get_raw, - redis_set, -) +from app.api.utils import SSE, broadcast_sse_msg, redis_get, redis_get_raw, redis_set from app.apps.constants import AppsServiceActions, AppsServiceKeys from app.apps.models import AppStatusQueryResult from app.external.result_type.src.result.result import Err, Ok, Result @@ -265,78 +258,3 @@ async def get_cache_timestamp( return Err( Report(message="Error retrieving cache timestamp from Redis", error=e) ) - - -async def get_lock_status(redis: Redis | None = None) -> Result[bool, Report]: - """ - Checks if the update lock is held. - Returns True if lock is held. - """ - logger.trace("get_lock_status()") - result = await redis_get_raw( - AppsServiceKeys.APP_STATUS_LOCK_KEY, custom_redis=redis - ) - match result: - case Ok(None): - logger.debug("App status update lock not held.") - return Ok(False) - case Ok(data) if isinstance(data, bytes) and data.decode("utf-8") == "locked": - logger.debug("App status update lock held.") - return Ok(True) - case Ok(data) if not isinstance(data, bytes): - return Err(Report(message=f"Unexpected data from Redis: {data}")) - case Err(e): - logger.error("Error checking if app status update lock is held: %s", e) - return Err(e) - - return Err(Report(message="Error checking if app status update lock is held")) - - -async def acquire_update_lock(redis: Redis | None = None) -> Result[bool, Report]: - """ - Attempts to acquire a lock to prevent concurrent updates. - Returns True if lock acquired. - """ - logger.trace("acquire_update_lock()") - - result = await get_lock_status(redis) - match result: - case Ok(lock_status): - if lock_status: - return Ok(False) - case Err(report): - return Err(report) - - # nx: Only set the key if it does not already exist. - # ex: Set the specified expire time, in seconds. - match await redis_set( - AppsServiceKeys.APP_STATUS_LOCK_KEY, - "locked", - nx=True, - ex=LOCK_TTL_SECONDS, - custom_redis=redis, - ): - case Ok(_): - return Ok(True) - case Err(report): - return Err( - report.attach_frame( - Frame(message="Error acquiring update lock in Redis") - ) - ) - - return Err(Report(message="Error acquiring update lock in Redis")) - - -async def release_update_lock(redis: Redis | None = None) -> Result[None, Report]: - """Releases the update lock.""" - logger.trace("release_update_lock()") - match await redis_delete(AppsServiceKeys.APP_STATUS_LOCK_KEY, custom_redis=redis): - case Ok(_): - return Ok(None) - case Err(report): - return Err( - report.attach_frame( - Frame(message="Error releasing update lock in Redis") - ) - ) diff --git a/app/apps/tasks.py b/app/apps/tasks.py index ae381cd..956b0d4 100644 --- a/app/apps/tasks.py +++ b/app/apps/tasks.py @@ -1,54 +1,9 @@ -""" -This module defines a Celery task `update_app_state_task` responsible for updating the -application state cache and notifying changes via a Redis channel. The task performs -the following operations: - -1. Checks if a Redis client is available and logs an error if not. -2. Attempts to acquire a lock to ensure only one instance of the task runs at a time. -3. Connects to a Redis channel for notifications. -4. Fetches the latest application status using the platform-specific implementation. -5. Updates the application status cache in Redis. -6. Notifies the Redis channel about the status update or any errors encountered. -7. Releases the lock after the task is completed, ensuring proper cleanup. - -The task handles various error scenarios, logging detailed error messages and notifying -the Redis channel about failures. -""" - import asyncio -import json -from typing import Optional -from loguru import logger -from redis.asyncio import Redis -from redis.asyncio import from_url as redis_from_url - -from app.api.channel import BaseChannelNotifier from app.api.config import config -from app.api.error_report.report import Report -from app.api.models import ApiErrors, ErrorMessage -from app.apps.cache import ( - acquire_update_lock, - release_update_lock, - set_cached_app_status, -) -from app.apps.constants import AppsServiceActions, AppsServiceKeys +from app.apps.models import AppId +from app.apps.tasks_impl import update_app_state_task_impl from app.celery_app import celery_app -from app.external.result_type.src.result import Err, Ok, Result -from app.system.models import APIPlatform - -PLATFORM = config("BAPI_PLATFORM", default=APIPlatform.UNKNOWN) -if PLATFORM == APIPlatform.RASPIBLITZ: - from app.apps.impl.raspiblitz import RaspiBlitzApps as Apps -elif PLATFORM == APIPlatform.NATIVE_PYTHON: - from app.apps.impl.native_python import NativePythonApps as Apps -else: - raise RuntimeError( - f"Unsupported platform '{PLATFORM}'. Options: {APIPlatform.values_as_list()}." - ) - - -apps_impl = Apps() BAPI_REDIS_URL = config("BAPI_REDIS_URL", "redis://127.0.0.1:6379/0") @@ -62,180 +17,6 @@ def update_app_state_task(): Celery task to fetch the latest app status, update the cache and notify on Redis channel. """ - asyncio.run(_update_app_state_task()) + asyncio.run(update_app_state_task_impl(BAPI_REDIS_URL)) -async def _update_app_state_task(): - redis_client = None - try: - redis_client = redis_from_url(BAPI_REDIS_URL, decode_responses=False) - except Exception as e: - logger.error(f"Failed to initialize Redis client: {e}") - logger.info("Attempting to run update_app_state_task...") - - if not isinstance(redis_client, Redis): - raise Exception("Redis not properly initialized, got a Sentinel") - - result = await acquire_update_lock(redis_client) - match result: - case Ok(acquired) if not acquired: - logger.warning("App state update lock already held. Skipping task run.") - return await redis_client.close() - case Err(report): - logger.error(f"Failed to acquire update lock: {report.format_verbose()}") - return await redis_client.close() - - channel_notifier = BaseChannelNotifier( - AppsServiceKeys.APP_STATE_CHANNEL, redis_url=BAPI_REDIS_URL - ) - result = await channel_notifier.connect() - match result: - case Ok(_): - logger.info("App state channel connected.") - case Err(report): - logger.error( - f"Failed to connect to app state channel: {report.format_verbose()}" - ) - return await redis_client.close() - - try: - logger.info("App state update lock acquired. Fetching app status...") - - await channel_notifier.notify_key_change( - AppsServiceKeys.APP_STATUS_CACHE_KEY, AppsServiceActions.STARTED - ) - - result = await apps_impl.get_app_status() - match result: - case Ok(data): - logger.info("Successfully fetched app status.") - res = await set_cached_app_status(data, redis_client) - match res: - case Ok(_): - res = await channel_notifier.notify_key_change( - AppsServiceKeys.APP_STATUS_CACHE_KEY, - AppsServiceActions.UPDATED, - None, - data.model_dump_json(), - ) - _handle_result(res) - case Err(report): - logger.error( - f"Failed to update app status cache: " - f"{report.format_verbose()}" - ) - await _handle_task_error( - channel_notifier, - AppsServiceKeys.APP_STATUS_UPDATE_FAILED_KEY, - AppsServiceActions.ERROR, - f"Failed to update app status cache: " - f"{report.frames[0].message}", - ApiErrors.APP_STATUS_UPDATE_FAILED, - report, - ) - case Err(report): - logger.error(f"Failed to fetch app status: {report.format()}") - await _handle_task_error( - channel_notifier, - AppsServiceKeys.APP_STATUS_UPDATE_FAILED_KEY, - AppsServiceActions.ERROR, - f"Failed to update app status: {report.frames[0].message}", - ApiErrors.APP_STATUS_UPDATE_FAILED, - report, - ) - - except Exception as e: - logger.exception(f"Unexpected error during update_app_state_task: {e}") - error_report = Report( - f"Unexpected error during app status update: {str(e)}", error=e - ) - await _handle_task_error( - channel_notifier, - AppsServiceKeys.APP_STATUS_UPDATE_FAILED_KEY, - AppsServiceActions.ERROR, - f"Unexpected error during app status update: {str(e)}", - ApiErrors.BACKGROUND_TASK_FAILED, - error_report, - ) - - finally: - res = await release_update_lock(redis=redis_client) - match res: - case Ok(_): - logger.debug("App state update lock released.") - case Err(report): - logger.error( - f"Failed to release app status update lock: {report.format()}" - ) - await _handle_task_error( - channel_notifier, - AppsServiceKeys.APP_STATUS_LOCK_KEY, - AppsServiceActions.LOCK_ERROR, - "Failed to release app status update lock:" - f" {report.frames[0].message}", - ApiErrors.APP_STATUS_UPDATE_FAILED, - report, - ) - logger.debug("App state update lock released.") - - if redis_client: - await redis_client.close() - - -def _handle_result(res): - """Simple helper to log result of operations""" - match res: - case Ok(_): - logger.info("App status cache updated.") - case Err(report): - logger.error( - f"Failed to update app status cache: {report.format_verbose()}" - ) - - -async def _handle_task_error( - channel_notifier: BaseChannelNotifier, - key: str, - action: str, - error_message: str, - error_code: str, - report: Optional[Report] = None, -) -> Result[None, Report]: - """Centralized error handling for tasks with channel notification - - Parameters - ---------- - channel_notifier : BaseChannelNotifier - The channel notifier to use for sending the error notification - key : str - The key to use for the notification - action : str - The action to use for the notification - error_message : str - The error message to include in the notification - error_code : str - The error code to include in the notification - report : Optional[Report] - The error report to include in the notification, if any - - Returns - ------- - Result[None, Report] - The result of the notification operation - """ - try: - error_payload = ErrorMessage( - detail=error_message, - error_code=error_code, - report=report.format_verbose() if report else None, - ).model_dump() - - return await channel_notifier.notify_key_change( - key, - action, - None, - json.dumps(error_payload), - ) - except Exception as e: - logger.exception(f"Error handling task error: {e}") - return Err(Report(f"Error handling task error: {e}", error=e)) diff --git a/app/apps/tasks_impl/__init__.py b/app/apps/tasks_impl/__init__.py new file mode 100644 index 0000000..f7c6b55 --- /dev/null +++ b/app/apps/tasks_impl/__init__.py @@ -0,0 +1 @@ +from .app_status_update import update_app_state_task_impl diff --git a/app/apps/tasks_impl/app_status_update.py b/app/apps/tasks_impl/app_status_update.py new file mode 100644 index 0000000..81fc3e1 --- /dev/null +++ b/app/apps/tasks_impl/app_status_update.py @@ -0,0 +1,209 @@ +import json +from typing import Optional + +from loguru import logger +from redis.asyncio import Redis +from redis.asyncio import from_url as redis_from_url + +from app.api.channel import BaseChannelNotifier +from app.api.config import config +from app.api.error_report.report import Report +from app.api.models import ApiErrors, ErrorMessage +from app.api.task_utils import acquire_lock, release_update_lock +from app.apps.cache import LOCK_TTL_SECONDS, set_cached_app_status +from app.apps.constants import AppsServiceActions, AppsServiceKeys +from app.external.result_type.src.result import Err, Ok, Result +from app.system.models import APIPlatform + +PLATFORM = config("BAPI_PLATFORM", default=APIPlatform.UNKNOWN) +if PLATFORM == APIPlatform.RASPIBLITZ: + from app.apps.impl.raspiblitz import RaspiBlitzApps as Apps +elif PLATFORM == APIPlatform.NATIVE_PYTHON: + from app.apps.impl.native_python import NativePythonApps as Apps +else: + raise RuntimeError( + f"Unsupported platform '{PLATFORM}'. Options: {APIPlatform.values_as_list()}." + ) + + +async def update_app_state_task_impl(redis_url: str): + redis_client = None + try: + redis_client = redis_from_url(redis_url, decode_responses=False) + except Exception as e: + logger.error(f"Failed to initialize Redis client: {e}") + logger.info("Attempting to run update_app_state_task...") + + if not isinstance(redis_client, Redis): + raise Exception("Redis not properly initialized, got a Sentinel") + + result = await acquire_lock( + key=AppsServiceKeys.APP_STATUS_LOCK_KEY, + lock_ttl=LOCK_TTL_SECONDS, + redis=redis_client, + ) + match result: + case Ok(acquired) if not acquired: + logger.warning("App state update lock already held. Skipping task run.") + return await redis_client.close() + case Err(report): + logger.error(f"Failed to acquire update lock: {report.format_verbose()}") + return await redis_client.close() + + channel_notifier = BaseChannelNotifier( + AppsServiceKeys.APP_STATE_CHANNEL, redis_url=redis_url + ) + result = await channel_notifier.connect() + match result: + case Ok(_): + logger.info("App state channel connected.") + case Err(report): + logger.error( + f"Failed to connect to app state channel: {report.format_verbose()}" + ) + return await redis_client.close() + + try: + logger.info("App state update lock acquired. Fetching app status...") + + await channel_notifier.notify_key_change( + AppsServiceKeys.APP_STATUS_CACHE_KEY, AppsServiceActions.STARTED + ) + + apps_impl = Apps() + result = await apps_impl.get_app_status() + match result: + case Ok(data): + logger.info("Successfully fetched app status.") + res = await set_cached_app_status(data, redis_client) + match res: + case Ok(_): + res = await channel_notifier.notify_key_change( + AppsServiceKeys.APP_STATUS_CACHE_KEY, + AppsServiceActions.UPDATED, + None, + data.model_dump_json(), + ) + _handle_result(res) + case Err(report): + logger.error( + f"Failed to update app status cache: " + f"{report.format_verbose()}" + ) + await _handle_task_error( + channel_notifier, + AppsServiceKeys.APP_STATUS_UPDATE_FAILED_KEY, + AppsServiceActions.ERROR, + f"Failed to update app status cache: " + f"{report.frames[0].message}", + ApiErrors.APP_STATUS_UPDATE_FAILED, + report, + ) + case Err(report): + logger.error(f"Failed to fetch app status: {report.format()}") + await _handle_task_error( + channel_notifier, + AppsServiceKeys.APP_STATUS_UPDATE_FAILED_KEY, + AppsServiceActions.ERROR, + f"Failed to update app status: {report.frames[0].message}", + ApiErrors.APP_STATUS_UPDATE_FAILED, + report, + ) + + except Exception as e: + logger.exception(f"Unexpected error during update_app_state_task: {e}") + error_report = Report( + f"Unexpected error during app status update: {str(e)}", error=e + ) + await _handle_task_error( + channel_notifier, + AppsServiceKeys.APP_STATUS_UPDATE_FAILED_KEY, + AppsServiceActions.ERROR, + f"Unexpected error during app status update: {str(e)}", + ApiErrors.BACKGROUND_TASK_FAILED, + error_report, + ) + + finally: + res = await release_update_lock( + key=AppsServiceKeys.APP_STATUS_LOCK_KEY, redis=redis_client + ) + match res: + case Ok(_): + logger.debug("App state update lock released.") + case Err(report): + logger.error( + f"Failed to release app status update lock: {report.format()}" + ) + await _handle_task_error( + channel_notifier, + AppsServiceKeys.APP_STATUS_LOCK_KEY, + AppsServiceActions.LOCK_ERROR, + "Failed to release app status update lock:" + f" {report.frames[0].message}", + ApiErrors.APP_STATUS_UPDATE_FAILED, + report, + ) + logger.debug("App state update lock released.") + + if redis_client: + await redis_client.close() + + +def _handle_result(res): + """Simple helper to log result of operations""" + match res: + case Ok(_): + logger.info("App status cache updated.") + case Err(report): + logger.error( + f"Failed to update app status cache: {report.format_verbose()}" + ) + + +async def _handle_task_error( + channel_notifier: BaseChannelNotifier, + key: str, + action: str, + error_message: str, + error_code: str, + report: Optional[Report] = None, +) -> Result[None, Report]: + """Centralized error handling for tasks with channel notification + + Parameters + ---------- + channel_notifier : BaseChannelNotifier + The channel notifier to use for sending the error notification + key : str + The key to use for the notification + action : str + The action to use for the notification + error_message : str + The error message to include in the notification + error_code : str + The error code to include in the notification + report : Optional[Report] + The error report to include in the notification, if any + + Returns + ------- + Result[None, Report] + The result of the notification operation + """ + try: + error_payload = ErrorMessage( + detail=error_message, + error_code=error_code, + report=report.format_verbose() if report else None, + ).model_dump() + + return await channel_notifier.notify_key_change( + key, + action, + None, + json.dumps(error_payload), + ) + except Exception as e: + logger.exception(f"Error handling task error: {e}") + return Err(Report(f"Error handling task error: {e}", error=e))