fix: harden ARQ cancellation and document upload retries

- enable ARQ job aborts via WorkerSettings
- add explicit Telegram document read/write timeouts
- stop network-failed document sends from retrying forever

Author: deadman96385

dumpyarabot/arq_config.py

@@ -56,11 +56,12 @@ def get_functions(cls):
     # Retry configuration
     max_tries = 3
 
-    # Health check
-    health_check_interval = 30
-
-    # Logging
-    log_results = True
+    # Health check
+    health_check_interval = 30
+    allow_abort_jobs = True
+
+    # Logging
+    log_results = True
 
     # Result TTL configuration (seconds)
     result_ttl = {

dumpyarabot/config.py

@@ -21,6 +21,8 @@ class Settings(BaseSettings):
 
     # Telegram formatting configuration
     DEFAULT_PARSE_MODE: str = "Markdown"
+    TELEGRAM_DOCUMENT_READ_TIMEOUT: float = 120.0
+    TELEGRAM_DOCUMENT_WRITE_TIMEOUT: float = 120.0
 
     # Optional custom base URL for Telegram Bot API (e.g. nginx reverse proxy)
     # Default: https://api.telegram.org/bot

dumpyarabot/message_queue.py

@@ -451,6 +451,8 @@ async def _process_message(self, message: QueuedMessage) -> bool:
                     ),
                     caption=message.caption,
                     parse_mode=message.parse_mode,
+                    read_timeout=settings.TELEGRAM_DOCUMENT_READ_TIMEOUT,
+                    write_timeout=settings.TELEGRAM_DOCUMENT_WRITE_TIMEOUT,
                 )
                 console.print(f"[green]Successfully processed {message.type.value} message[/green]")
                 return True
@@ -510,13 +512,24 @@ async def _process_message(self, message: QueuedMessage) -> bool:
             await self._requeue_message(message)
             return True  # Don't increment retry count for rate limits
 
-        except NetworkError as e:
-            console.print(f"[yellow]Network error processing message: {e}[/yellow]")
-            # Simple retry with exponential backoff for network issues
-            retry_delay = min(30 * (2 ** message.retry_count), 300)  # 30s, 60s, 120s, 240s, 300s max
-            message.scheduled_for = datetime.now(timezone.utc) + timedelta(seconds=retry_delay)
-            await self._requeue_message(message)
-            return True  # Don't increment retry count for network issues
+        except NetworkError as e:
+            console.print(f"[yellow]Network error processing message: {e}[/yellow]")
+            message.retry_count += 1
+            if message.retry_count <= message.max_retries:
+                retry_delay = min(30 * (2 ** (message.retry_count - 1)), 300)
+                console.print(
+                    f"[yellow]Retrying message {message.message_id} after network error "
+                    f"(attempt {message.retry_count}/{message.max_retries}) in {retry_delay}s[/yellow]"
+                )
+                message.scheduled_for = datetime.now(timezone.utc) + timedelta(seconds=retry_delay)
+                await self._requeue_message(message)
+            else:
+                console.print(
+                    f"[red]Message {message.message_id} exceeded max retries after network errors, "
+                    f"moving to dead letter queue[/red]"
+                )
+                await self._move_to_dead_letter_queue(message)
+            return True
 
         except TelegramError as e:
             console.print(f"[red]Telegram API error processing message: {e}[/red]")

run_arq_worker.py

@@ -48,6 +48,7 @@ async def start_worker(self):
                 keep_result=WorkerSettings.keep_result,
                 max_tries=WorkerSettings.max_tries,
                 health_check_interval=WorkerSettings.health_check_interval,
+                allow_abort_jobs=WorkerSettings.allow_abort_jobs,
                 queue_name=WorkerSettings.queue_name
             )