From 4ef7b00c161b1f1a9d37bb67f8acbfadd2a2a38b Mon Sep 17 00:00:00 2001 From: vvzvlad Date: Mon, 16 Mar 2026 01:07:19 +0300 Subject: [PATCH] refactor(api): split download semaphores and remove diagnostic monitoring MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Separate live HTTP media requests and background cache workers by introducing `HTTP_DOWNLOAD_SEMAPHORE` and `BACKGROUND_DOWNLOAD_SEMAPHORE`. Remove the previous global `DOWNLOAD_SEMAPHORE`, diagnostic event‑loop monitor, and related logging. Refactor `download_media_file` to handle large videos with temporary files, clean up zero‑size cache entries, and update access timestamps without semaphore tracking. Adjust background worker and request handling to use the new semaphores, simplifying concurrency control and eliminating unnecessary diagnostics. --- api_server.py | 295 ++++++++++++++++++++++---------------------------- 1 file changed, 131 insertions(+), 164 deletions(-) diff --git a/api_server.py b/api_server.py index 97f2f84..f503a0d 100644 --- a/api_server.py +++ b/api_server.py @@ -67,29 +67,10 @@ if not logger.handlers: pass client = TelegramClient() Config = get_settings() -DOWNLOAD_SEMAPHORE = asyncio.Semaphore(3) -_semaphore_waiters = 0 # diagnostic: track how many requests wait for semaphore +HTTP_DOWNLOAD_SEMAPHORE = asyncio.Semaphore(3) # semaphore for live HTTP media requests +BACKGROUND_DOWNLOAD_SEMAPHORE = asyncio.Semaphore(2) # semaphore for background cache worker download_queue = asyncio.Queue(maxsize=100) -async def _diag_event_loop_monitor(): - """Periodic diagnostic: logs event loop tasks, semaphore state, and threadpool active threads.""" - import threading - while True: - await asyncio.sleep(30) - loop = asyncio.get_event_loop() - all_tasks = asyncio.all_tasks(loop) - task_names = [t.get_name() for t in all_tasks] - logger.warning( - f"[DIAG] event_loop_monitor: " - f"active_tasks={len(all_tasks)}, " - f"semaphore_value={DOWNLOAD_SEMAPHORE._value}, " - f"semaphore_waiters={_semaphore_waiters}, " - f"download_queue_size={download_queue.qsize()}, " - f"active_threads={threading.active_count()}, " - f"tasks={task_names}" - ) - - @asynccontextmanager async def lifespan(_: FastAPI): setup_logging(Config["log_level"]) @@ -101,11 +82,9 @@ async def lifespan(_: FastAPI): await client.start() background_task = asyncio.create_task(cache_media_files()) # Start background task worker_task = asyncio.create_task(background_download_worker()) # Start download worker - diag_task = asyncio.create_task(_diag_event_loop_monitor()) # Start diagnostic monitor yield background_task.cancel() # Cleanup worker_task.cancel() - diag_task.cancel() try: await background_task except asyncio.CancelledError: @@ -114,10 +93,6 @@ async def lifespan(_: FastAPI): await worker_task except asyncio.CancelledError: pass - try: - await diag_task - except asyncio.CancelledError: - pass await client.stop() app = FastAPI(title="Pyrogram Bridge", lifespan=lifespan) @@ -218,10 +193,7 @@ def delayed_delete_file(file_path: str, delay: int = 300) -> None: Delay is set to 5 minutes by default. NOTE: Runs in a background thread via Starlette BackgroundTask. """ - import threading - logger.warning(f"[DIAG] delayed_delete_file: starting sleep({delay}s) in thread {threading.current_thread().name}, active threads: {threading.active_count()}") time.sleep(delay) - logger.warning(f"[DIAG] delayed_delete_file: sleep finished, deleting {file_path}, active threads: {threading.active_count()}") try: os.remove(file_path) logger.info(f"Deleted temporary file {file_path} after delay of {delay} seconds") @@ -255,146 +227,140 @@ async def download_media_file(channel: Union[str, int], post_id: int, file_uniqu Download media file from Telegram and save to cache Returns tuple of (file path, delete_after) """ - global _semaphore_waiters - _semaphore_waiters += 1 - logger.warning(f"[DIAG] download_media_file: waiting for semaphore, waiters={_semaphore_waiters}, semaphore_value={DOWNLOAD_SEMAPHORE._value}, channel={channel}, post_id={post_id}") - async with DOWNLOAD_SEMAPHORE: - _semaphore_waiters -= 1 - logger.warning(f"[DIAG] download_media_file: acquired semaphore, waiters_remaining={_semaphore_waiters}, semaphore_value={DOWNLOAD_SEMAPHORE._value}, channel={channel}, post_id={post_id}") - base_cache_dir = os.path.abspath("./data/cache") - - # Create nested cache structure - channel_dir = os.path.join(base_cache_dir, str(channel)) - post_dir = os.path.join(channel_dir, str(post_id)) - os.makedirs(post_dir, exist_ok=True) - - # Convert numeric channel ID to int if needed - channel_id: Union[str, int] = channel - if isinstance(channel, str) and channel.startswith('-100'): - channel_id = int(channel) - + base_cache_dir = os.path.abspath("./data/cache") + + # Create nested cache structure + channel_dir = os.path.join(base_cache_dir, str(channel)) + post_dir = os.path.join(channel_dir, str(post_id)) + os.makedirs(post_dir, exist_ok=True) + + # Convert numeric channel ID to int if needed + channel_id: Union[str, int] = channel + if isinstance(channel, str) and channel.startswith('-100'): + channel_id = int(channel) + + try: + message = await client.safe_get_messages(channel_id, post_id) + except asyncio.TimeoutError: + logger.error(f"Timeout getting messages for {channel}/{post_id}") + raise HTTPException(status_code=504, detail="Request timeout") + + if message.media == "MessageMediaType.POLL": + return None, False + + # Check if it is a video and if its size exceeds 100 MB + is_large_video = False + if message.video: try: - message = await client.safe_get_messages(channel_id, post_id) - except asyncio.TimeoutError: - logger.error(f"Timeout getting messages for {channel}/{post_id}") - raise HTTPException(status_code=504, detail="Request timeout") - - if message.media == "MessageMediaType.POLL": - return None, False - - # Check if it is a video and if its size exceeds 100 MB - is_large_video = False - if message.video: - try: - if message.video.file_size > 100 * 1024 * 1024: - is_large_video = True - except Exception as e: - logger.error(f"Failed to get video file size for message {post_id}: {str(e)}") - - if is_large_video: - # For large video, download without permanent caching; use a temporary file - temp_file_path = os.path.join(post_dir, f"temp_{file_unique_id}") - if os.path.exists(temp_file_path): - logger.info(f"Temporary file {temp_file_path} already exists, serving cached large video") - return temp_file_path, False - file_id = await find_file_id_in_message(message, file_unique_id) - if not file_id: - logger.error(f"Media with file_unique_id {file_unique_id} not found in message {post_id}") - raise HTTPException(status_code=404, detail="File not found in message") - logger.info(f"Downloading large video file {file_unique_id} to temporary path {temp_file_path}") - try: - file_path = await client.safe_download_media(file_id, temp_file_path) - except asyncio.TimeoutError: - logger.error(f"Timeout downloading large video {file_unique_id}") - raise HTTPException(status_code=504, detail="Download timeout") - logger.info(f"Downloaded large video file {file_unique_id} to temporary path {temp_file_path}") - return file_path, False - - # Normal caching flow - cache_path = os.path.join(post_dir, file_unique_id) - if os.path.exists(cache_path): - # Check if the cached file is zero size - if os.path.getsize(cache_path) == 0: - logger.warning(f"zero_size_cache_found: Found zero-size cached file: {cache_path}. Deleting and attempting redownload.") - try: - os.remove(cache_path) - logger.info(f"Removed zero-size cached file: {cache_path}") - except OSError as e: - logger.error(f"cleanup_error: Failed to remove zero-size cached file {cache_path}: {e}") - # Do not raise error here, proceed to download below - else: - # File exists and is not zero size, update timestamp and return - logger.info(f"Found cached media file: {cache_path}") - # Update access timestamp in media_file_ids.json - file_ids_path = os.path.join(os.path.abspath("./data"), 'media_file_ids.json') - try: - if os.path.exists(file_ids_path): - media_files = await asyncio.to_thread(read_json_file_sync, file_ids_path) - for file_data in media_files: - if (file_data.get('channel') == str(channel) and # Ensure channel is string for comparison - file_data.get('post_id') == post_id and - file_data.get('file_unique_id') == file_unique_id): - file_data['added'] = datetime.now().timestamp() - break - async with get_media_file_ids_lock(): - await asyncio.to_thread(write_json_file_sync, file_ids_path, media_files) - except Exception as e: - logger.error(f"Failed to update timestamp for {channel}/{post_id}/{file_unique_id}: {str(e)}") - return cache_path, False - + if message.video.file_size > 100 * 1024 * 1024: + is_large_video = True + except Exception as e: + logger.error(f"Failed to get video file size for message {post_id}: {str(e)}") + + if is_large_video: + # For large video, download without permanent caching; use a temporary file + temp_file_path = os.path.join(post_dir, f"temp_{file_unique_id}") + if os.path.exists(temp_file_path): + logger.info(f"Temporary file {temp_file_path} already exists, serving cached large video") + return temp_file_path, False file_id = await find_file_id_in_message(message, file_unique_id) if not file_id: - error_message = f"Media with file_unique_id {file_unique_id} not found in message {post_id} for channel {channel}" - logger.error(error_message) - - # Attempt to remove the invalid entry from media_file_ids.json using threadpool I/O + logger.error(f"Media with file_unique_id {file_unique_id} not found in message {post_id}") + raise HTTPException(status_code=404, detail="File not found in message") + logger.info(f"Downloading large video file {file_unique_id} to temporary path {temp_file_path}") + try: + file_path = await client.safe_download_media(file_id, temp_file_path) + except asyncio.TimeoutError: + logger.error(f"Timeout downloading large video {file_unique_id}") + raise HTTPException(status_code=504, detail="Download timeout") + logger.info(f"Downloaded large video file {file_unique_id} to temporary path {temp_file_path}") + return file_path, False + + # Normal caching flow + cache_path = os.path.join(post_dir, file_unique_id) + if os.path.exists(cache_path): + # Check if the cached file is zero size + if os.path.getsize(cache_path) == 0: + logger.warning(f"zero_size_cache_found: Found zero-size cached file: {cache_path}. Deleting and attempting redownload.") + try: + os.remove(cache_path) + logger.info(f"Removed zero-size cached file: {cache_path}") + except OSError as e: + logger.error(f"cleanup_error: Failed to remove zero-size cached file {cache_path}: {e}") + # Do not raise error here, proceed to download below + else: + # File exists and is not zero size, update timestamp and return + logger.info(f"Found cached media file: {cache_path}") + # Update access timestamp in media_file_ids.json file_ids_path = os.path.join(os.path.abspath("./data"), 'media_file_ids.json') try: if os.path.exists(file_ids_path): media_files = await asyncio.to_thread(read_json_file_sync, file_ids_path) - initial_count = len(media_files) - media_files_updated = [ - f_data for f_data in media_files - if not ( - f_data.get('channel') == str(channel) and - f_data.get('post_id') == post_id and - f_data.get('file_unique_id') == file_unique_id - ) - ] - if len(media_files_updated) < initial_count: - async with get_media_file_ids_lock(): - await asyncio.to_thread(write_json_file_sync, file_ids_path, media_files_updated) - logger.info(f"Removed invalid entry for {channel}/{post_id}/{file_unique_id} from media_file_ids.json") - else: - logger.warning(f"Entry for {channel}/{post_id}/{file_unique_id} not found in media_file_ids.json for removal") - + for file_data in media_files: + if (file_data.get('channel') == str(channel) and # Ensure channel is string for comparison + file_data.get('post_id') == post_id and + file_data.get('file_unique_id') == file_unique_id): + file_data['added'] = datetime.now().timestamp() + break + async with get_media_file_ids_lock(): + await asyncio.to_thread(write_json_file_sync, file_ids_path, media_files) except Exception as e: - logger.error(f"Failed to remove entry for {channel}/{post_id}/{file_unique_id} from media_file_ids.json: {str(e)}") - - raise HTTPException(status_code=404, detail="File not found in message") - - try: - file_path = await client.safe_download_media(file_id, cache_path) - except asyncio.TimeoutError: - logger.error(f"Timeout downloading media {file_unique_id}") - raise HTTPException(status_code=504, detail="Download timeout") - - # Check if the downloaded file exists and has a size greater than zero - if not file_path or not os.path.exists(file_path) or os.path.getsize(file_path) == 0: - logger.error(f"download_failed_zero_size: Downloaded file {file_unique_id} for {channel}/{post_id} is zero size or missing.") - # Attempt to clean up the invalid file - if file_path and os.path.exists(file_path): - try: - os.remove(file_path) - logger.info(f"Removed zero-size file: {file_path}") - except OSError as e: - logger.error(f"cleanup_error: Failed to remove zero-size file {file_path}: {e}") - # Raise an error to indicate download failure - # Raise specific error - raise ZeroSizeFileError(f"Downloaded file {file_unique_id} for {channel}/{post_id} is zero size or missing after download attempt.") + logger.error(f"Failed to update timestamp for {channel}/{post_id}/{file_unique_id}: {str(e)}") + return cache_path, False - logger.info(f"Downloaded media file {file_unique_id} to {cache_path}") - return file_path, False + file_id = await find_file_id_in_message(message, file_unique_id) + if not file_id: + error_message = f"Media with file_unique_id {file_unique_id} not found in message {post_id} for channel {channel}" + logger.error(error_message) + + # Attempt to remove the invalid entry from media_file_ids.json using threadpool I/O + file_ids_path = os.path.join(os.path.abspath("./data"), 'media_file_ids.json') + try: + if os.path.exists(file_ids_path): + media_files = await asyncio.to_thread(read_json_file_sync, file_ids_path) + initial_count = len(media_files) + media_files_updated = [ + f_data for f_data in media_files + if not ( + f_data.get('channel') == str(channel) and + f_data.get('post_id') == post_id and + f_data.get('file_unique_id') == file_unique_id + ) + ] + if len(media_files_updated) < initial_count: + async with get_media_file_ids_lock(): + await asyncio.to_thread(write_json_file_sync, file_ids_path, media_files_updated) + logger.info(f"Removed invalid entry for {channel}/{post_id}/{file_unique_id} from media_file_ids.json") + else: + logger.warning(f"Entry for {channel}/{post_id}/{file_unique_id} not found in media_file_ids.json for removal") + + except Exception as e: + logger.error(f"Failed to remove entry for {channel}/{post_id}/{file_unique_id} from media_file_ids.json: {str(e)}") + + raise HTTPException(status_code=404, detail="File not found in message") + + try: + file_path = await client.safe_download_media(file_id, cache_path) + except asyncio.TimeoutError: + logger.error(f"Timeout downloading media {file_unique_id}") + raise HTTPException(status_code=504, detail="Download timeout") + + # Check if the downloaded file exists and has a size greater than zero + if not file_path or not os.path.exists(file_path) or os.path.getsize(file_path) == 0: + logger.error(f"download_failed_zero_size: Downloaded file {file_unique_id} for {channel}/{post_id} is zero size or missing.") + # Attempt to clean up the invalid file + if file_path and os.path.exists(file_path): + try: + os.remove(file_path) + logger.info(f"Removed zero-size file: {file_path}") + except OSError as e: + logger.error(f"cleanup_error: Failed to remove zero-size file {file_path}: {e}") + # Raise an error to indicate download failure + # Raise specific error + raise ZeroSizeFileError(f"Downloaded file {file_unique_id} for {channel}/{post_id} is zero size or missing after download attempt.") + + logger.info(f"Downloaded media file {file_unique_id} to {cache_path}") + return file_path, False def remove_old_cached_files_sync(media_files: list, cache_dir: str) -> tuple[list, int]: @@ -556,10 +522,10 @@ async def background_download_worker(): try: channel, post_id, file_unique_id = await download_queue.get() logger.info(f"Background download: {channel}/{post_id}/{file_unique_id}") - logger.warning(f"[DIAG] background_download_worker: queue_size={download_queue.qsize()}, semaphore_value={DOWNLOAD_SEMAPHORE._value}, waiters={_semaphore_waiters}") try: - await download_media_file(channel, post_id, file_unique_id) + async with BACKGROUND_DOWNLOAD_SEMAPHORE: # limit concurrent background downloads + await download_media_file(channel, post_id, file_unique_id) await asyncio.sleep(2) except Exception as e: logger.error(f"Background download error for {channel}/{post_id}/{file_unique_id}: {e}") @@ -823,7 +789,8 @@ async def get_media(channel: str, post_id: int, file_unique_id: str, digest: str channel_id = int(channel) try: # Wrap the download and prepare call - file_path, delete_after = await download_media_file(channel_id, post_id, file_unique_id) + async with HTTP_DOWNLOAD_SEMAPHORE: # limit concurrent live HTTP downloads + file_path, delete_after = await download_media_file(channel_id, post_id, file_unique_id) if not file_path: raise HTTPException(status_code=404, detail="File not found") if file_path: