simplify queue ops, handle broken pipe errors
- Add OSError errno 32 (broken pipe) handling in progress broadcast - Remove progress service calls from add/clear queue operations - Add pending_by_episode cleanup on clear - Update tests accordingly
This commit is contained in:
@@ -229,7 +229,7 @@ async def clear_pending(
|
||||
)
|
||||
|
||||
|
||||
@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
@router.delete("/{item_id}", status_code=status.HTTP_200_OK)
|
||||
async def remove_from_queue(
|
||||
item_id: str = Path(..., description="Download item ID to remove"),
|
||||
_: dict = Depends(require_auth),
|
||||
@@ -246,6 +246,9 @@ async def remove_from_queue(
|
||||
Args:
|
||||
item_id: Unique identifier of the download item to remove
|
||||
|
||||
Returns:
|
||||
dict: Status message confirming removal
|
||||
|
||||
Raises:
|
||||
HTTPException: 401 if not authenticated, 404 if item not found,
|
||||
500 on service error
|
||||
@@ -260,6 +263,12 @@ async def remove_from_queue(
|
||||
resource_id=item_id
|
||||
)
|
||||
|
||||
return {
|
||||
"status": "success",
|
||||
"message": f"Removed item {item_id} from queue",
|
||||
"removed_id": item_id,
|
||||
}
|
||||
|
||||
except DownloadServiceError as e:
|
||||
raise BadRequestError(message=str(e))
|
||||
except (BadRequestError, NotFoundError, ServerError):
|
||||
|
||||
@@ -532,6 +532,15 @@ class DownloadService:
|
||||
"Queue progress already initialized by concurrent task"
|
||||
)
|
||||
self._queue_progress_initialized = True
|
||||
# Handle broken pipe / connection errors from WebSocket broadcast
|
||||
# These are non-fatal and should not fail the queue operation
|
||||
elif isinstance(e, OSError) and e.errno == 32:
|
||||
logger.warning(
|
||||
"Queue progress broadcast failed (broken pipe) - "
|
||||
"continuing without progress tracking: %s",
|
||||
e,
|
||||
)
|
||||
self._queue_progress_initialized = True
|
||||
else:
|
||||
logger.error("Failed to initialize queue progress: %s", e)
|
||||
|
||||
@@ -674,17 +683,10 @@ class DownloadService:
|
||||
episode=episode.episode,
|
||||
)
|
||||
|
||||
# Notify via progress service
|
||||
queue_status = await self.get_queue_status()
|
||||
await self._progress_service.update_progress(
|
||||
progress_id="download_queue",
|
||||
message=f"Added {len(created_ids)} items to queue",
|
||||
metadata={
|
||||
"action": "items_added",
|
||||
"added_ids": created_ids,
|
||||
"queue_status": queue_status.model_dump(mode="json"),
|
||||
},
|
||||
force_broadcast=True,
|
||||
logger.info(
|
||||
"Added items to queue",
|
||||
count=len(created_ids),
|
||||
serie_key=serie_id,
|
||||
)
|
||||
|
||||
return created_ids
|
||||
@@ -1086,27 +1088,17 @@ class DownloadService:
|
||||
"""
|
||||
count = len(self._pending_queue)
|
||||
|
||||
# Delete all pending items from database
|
||||
for item_id in list(self._pending_items_by_id.keys()):
|
||||
await self._delete_from_database(item_id)
|
||||
# Only try to delete from DB if there are items
|
||||
if count > 0:
|
||||
for item_id in list(self._pending_items_by_id.keys()):
|
||||
try:
|
||||
await self._delete_from_database(item_id)
|
||||
except Exception as e:
|
||||
logger.warning("Failed to delete item %s: %s", item_id, e)
|
||||
|
||||
self._pending_queue.clear()
|
||||
self._pending_items_by_id.clear()
|
||||
logger.info("Cleared pending items count=%s", count)
|
||||
|
||||
# Notify via progress service
|
||||
if count > 0:
|
||||
queue_status = await self.get_queue_status()
|
||||
await self._progress_service.update_progress(
|
||||
progress_id="download_queue",
|
||||
message=f"Cleared {count} pending items",
|
||||
metadata={
|
||||
"action": "pending_cleared",
|
||||
"cleared_count": count,
|
||||
"queue_status": queue_status.model_dump(mode="json"),
|
||||
},
|
||||
force_broadcast=True,
|
||||
)
|
||||
self._pending_by_episode.clear()
|
||||
|
||||
return count
|
||||
|
||||
|
||||
Reference in New Issue
Block a user