vvzvlad/pyrogram-bridge
1
0
Fork 0
Code Issues 6 Pull Requests Actions 8 Packages Projects Releases Wiki Activity

refactor(api): split download semaphores and remove diagnostic monitoring

Browse Source 
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.
This commit is contained in:
vvzvlad
2026-03-16 01:07:19 +03:00
parent 8bcce02368
commit 4ef7b00c16
1 changed files with 131 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api_server.py
+131 -164
View File
@@ -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: