From becd32f54977ba85a498cb5ed58d7f704dc134a3 Mon Sep 17 00:00:00 2001 From: dgtlmoon Date: Mon, 5 Jan 2026 14:29:18 +0100 Subject: [PATCH] retry improvements --- .../blueprint/settings/__init__.py | 24 +- .../templates/failed-notifications.html | 260 ++++++ changedetectionio/notification/task_queue.py | 806 ++++++++++++++++-- 3 files changed, 1021 insertions(+), 69 deletions(-) create mode 100644 changedetectionio/blueprint/settings/templates/failed-notifications.html diff --git a/changedetectionio/blueprint/settings/__init__.py b/changedetectionio/blueprint/settings/__init__.py index 7d65f993e..b1c0e4b81 100644 --- a/changedetectionio/blueprint/settings/__init__.py +++ b/changedetectionio/blueprint/settings/__init__.py @@ -146,16 +146,20 @@ def construct_blueprint(datastore: ChangeDetectionStore): @login_optionally_required def failed_notifications(): """View notifications that failed all retry attempts""" - from changedetectionio.notification.task_queue import get_failed_notifications, get_retry_config, get_pending_notifications_count + from changedetectionio.notification.task_queue import get_failed_notifications, get_retry_config, get_pending_notifications_count, get_last_successful_notification, get_pending_notifications failed = get_failed_notifications(limit=100) retry_config = get_retry_config() pending_count = get_pending_notifications_count() + pending_list = get_pending_notifications(limit=50) + last_success = get_last_successful_notification() output = render_template("failed-notifications.html", failed_notifications=failed, retry_config=retry_config, - pending_count=pending_count) + pending_count=pending_count, + pending_list=pending_list, + last_success=last_success) return output @settings_blueprint.route("/retry-notification/", methods=['POST']) @@ -190,6 +194,22 @@ def construct_blueprint(datastore: ChangeDetectionStore): return redirect(url_for('settings.failed_notifications')) + @settings_blueprint.route("/clear-all-notifications", methods=['POST']) + @login_optionally_required + def clear_all_notifications(): + """Clear ALL notifications (queue, schedule, results, retry attempts)""" + from changedetectionio.notification.task_queue import clear_all_notifications as clear_all + + result = clear_all() + + if 'error' in result: + flash(f"Error clearing notifications: {result['error']}", 'error') + else: + total = result['queue'] + result['schedule'] + result['results'] + result['retry_attempts'] + result.get('task_metadata', 0) + flash(f"Cleared {total} notification(s): {result['queue']} queued, {result['schedule']} scheduled, {result['results']} failed, {result['retry_attempts']} retry attempts, {result.get('task_metadata', 0)} task metadata.", 'notice') + + return redirect(url_for('settings.failed_notifications')) + @settings_blueprint.route("/api/v1/notifications/failed", methods=['GET']) @login_optionally_required def api_get_failed_notifications(): diff --git a/changedetectionio/blueprint/settings/templates/failed-notifications.html b/changedetectionio/blueprint/settings/templates/failed-notifications.html new file mode 100644 index 000000000..aa7e57fe1 --- /dev/null +++ b/changedetectionio/blueprint/settings/templates/failed-notifications.html @@ -0,0 +1,260 @@ +{% extends 'base.html' %} + +{% block content %} +
+
+ +

Failed Notifications (Exhausted Retries)

+ + + {% if last_success %} +
+
✅ Most Recent Successful Notification
+
+
+ Timestamp: {{ last_success.timestamp_formatted }} +
+ {% if last_success.watch_url %} + + {% endif %} + {% if last_success.notification_urls %} +
+ Sent via: + {% for url in last_success.notification_urls %} + {{ url }} + {% endfor %} +
+ {% endif %} + {% if last_success.apprise_logs %} +
+ 📋 View Apprise Logs + 
{% for log_line in last_success.apprise_logs %}{{ log_line }}
+{% endfor %}
+
+ {% endif %} +
+

+ Use this as reference - this notification was sent successfully with current settings. +

+
+ {% endif %} + + +
+
Notification Queue Status
+
+
+ {% if pending_count is not none %} +
+ 🔄 Pending/Retrying: + {{ pending_count }} + notification{{ 's' if pending_count != 1 else '' }} +
+

+ Currently in queue or being retried +

+ {% else %} +
+ 🔄 Pending/Retrying: Unable to determine +
+ {% endif %} +
+
+
+ 💀 Failed (Dead Letter): + {{ failed_notifications|length }} + notification{{ 's' if failed_notifications|length != 1 else '' }} +
+

+ Exhausted all retry attempts +

+
+
+ + + {% if pending_list %} +
+ 📋 View Pending/Retrying Notifications ({{ pending_list|length }}) +
+ {% for item in pending_list %} +
+
+ {% if item.status == 'queued' %}⏳ Queued{% else %}🔄 Retrying{% endif %}: + {% if item.watch_url %} + {{ item.watch_url }} + {% else %} + Test notification + {% endif %} +
+ {% if item.status == 'retrying' and item.retry_at_formatted %} +
+ Retry at: {{ item.retry_at_formatted }} + {% if item.retry_in_seconds > 0 %} + (in {{ item.retry_in_seconds }}s) + {% endif %} +
+ {% endif %} +
+ {% endfor %} +
+
+ {% endif %} +
+ + +
+
Automatic Retry Schedule (Exponential Backoff)
+

+ Notifications are automatically retried {{ retry_config.retry_count }} times with exponential backoff starting at {{ retry_config.retry_delay_seconds }} seconds. +

+ + + + + + + + + + + + + + + {% for i in range(retry_config.retry_count) %} + {% set delay = retry_config.retry_delays[i] %} + {% set cumulative_time = retry_config.retry_delays[:i+1]|sum %} + + + + + + {% endfor %} + + + + + + +
AttemptTimeAction
1st (initial)T+0:00⚠️ Fails (e.g., SMTP server down)
{{ i + 2 }}{{ ['st', 'nd', 'rd'][i + 1] if i + 1 < 3 else 'th' }} (retry {{ i + 1 }})T+{{ '%d:%02d' % (cumulative_time // 60, cumulative_time % 60) }}{% if i < retry_config.retry_count - 1 %}⚠️ Fails → Wait {{ delay }}s ({{ '%d:%02d' % (delay // 60, delay % 60) }}){% else %}⚠️ Fails → Give up{% endif %}
Dead LetterT+{{ '%d:%02d' % (retry_config.total_time_seconds // 60, retry_config.total_time_seconds % 60) }}💀 Moved to this failed notifications list
+

+ Total: {{ retry_config.total_attempts }} attempts over {{ '%d:%02d' % (retry_config.total_time_seconds // 60, retry_config.total_time_seconds % 60) }} (mm:ss). + Failed notifications are kept for 30 days, then automatically deleted. +

+
+ + +
+ {% if failed_notifications|length > 0 %} +

+ These notifications failed after all {{ retry_config.retry_count }} retry attempts. + You can fix the notification settings (e.g., SMTP server) and retry them manually below. +

+
+ +
+ {% endif %} + + + {% if pending_count > 0 or failed_notifications|length > 0 %} +
+ +
+ {% endif %} + + {% if failed_notifications|length > 0 or pending_count > 0 %} +
+ {% if failed_notifications|length > 0 %} + Retry All: Re-queue failed notifications with current settings.
+ {% endif %} + Clear All: Delete ALL pending, retrying, and failed notifications (cannot be undone). +
+ {% endif %} +
+ + {% if failed_notifications|length == 0 %} +
+

+ ✅ No failed notifications - All notifications either succeeded or are still being retried. +

+
+ +
+ {% for notification in failed_notifications %} +
+ {% if notification.timestamp %} +
+ ⚠️ Failed: {{ notification.timestamp_formatted }} + {% if notification.days_ago is defined %} + ({{ notification.days_ago }} day{{ 's' if notification.days_ago != 1 else '' }} ago) + {% endif %} +
+ {% endif %} + +
+ Task ID: {{ notification.task_id }} +
+ + {% if notification.notification_data and notification.notification_data.get('watch_url') %} + + {% endif %} + + {% if notification.notification_data and notification.notification_data.get('uuid') %} +
+ Watch UUID: {{ notification.notification_data.get('uuid') }} +
+ {% endif %} + + {% if notification.retry_attempts %} +
+ Retry Attempts: +
+ {% for attempt in notification.retry_attempts %} +
+ + Attempt #{{ attempt.attempt_number }} - {{ attempt.timestamp_formatted }} + {% if attempt.will_retry %} + → Will retry + {% else %} + → Final attempt + {% endif %} + +
+ 
{{ attempt.error }}
+
+
+ {% endfor %} +
+
+ {% else %} +
+ Error: + 
{{ notification.error }}
+
+ {% endif %} + +
+ +
+
+ {% endfor %} +
+ {% endif %} + +
+
+ +{% endblock %} diff --git a/changedetectionio/notification/task_queue.py b/changedetectionio/notification/task_queue.py index ee474bd87..a47679565 100644 --- a/changedetectionio/notification/task_queue.py +++ b/changedetectionio/notification/task_queue.py @@ -12,6 +12,7 @@ Environment Variables: """ import os +import struct from loguru import logger # Get queue storage type from environment @@ -51,20 +52,23 @@ NOTIFICATION_RETRY_COUNT, NOTIFICATION_RETRY_DELAY = _get_retry_config() def get_retry_delays(): """ - Calculate retry delays with exponential backoff. + Calculate retry delays with exponential backoff for display purposes. - Returns a tuple of delays for each retry attempt. - Example: base delay 60s → (60, 120, 240, 480, ...) + Returns a list of delays for each retry attempt. + Example: base delay 60s → [60, 120, 240, 480, ...] + + Note: This is for display/reporting only. Actual exponential backoff + is handled by Huey's backoff=2 parameter in the task decorator. """ if NOTIFICATION_RETRY_COUNT == 0: - return tuple() + return [] delays = [] for i in range(NOTIFICATION_RETRY_COUNT): delay = NOTIFICATION_RETRY_DELAY * (2 ** i) # Exponential backoff delays.append(delay) - return tuple(delays) + return delays def get_retry_config(): @@ -176,12 +180,100 @@ def init_huey(datastore_path): return huey +def _count_storage_items(storage, storage_type): + """ + Count items in Huey storage (queue + schedule) based on storage backend type. + + Args: + storage: Huey storage instance + storage_type: Type name string (e.g., 'FileStorage', 'SqliteStorage', 'RedisStorage') + + Returns: + Tuple of (queue_count, schedule_count) + """ + queue_count = 0 + schedule_count = 0 + + import os + + if storage_type == 'FileStorage': + # FileStorage: Walk file directories + try: + if hasattr(storage, 'path'): + # Count queue files + queue_dir = os.path.join(storage.path, 'queue') + if os.path.exists(queue_dir): + for root, dirs, files in os.walk(queue_dir): + queue_count += len([f for f in files if not f.startswith('.')]) + + # Count schedule files + schedule_dir = os.path.join(storage.path, 'schedule') + if os.path.exists(schedule_dir): + for root, dirs, files in os.walk(schedule_dir): + schedule_count += len([f for f in files if not f.startswith('.')]) + except Exception as e: + logger.debug(f"FileStorage count error: {e}") + + elif storage_type in ['SqliteStorage', 'SqliteHuey']: + # SqliteStorage: Query database tables + try: + import sqlite3 + if hasattr(storage, 'filename'): + conn = sqlite3.connect(storage.filename) + cursor = conn.cursor() + + # Count queue + cursor.execute("SELECT COUNT(*) FROM queue") + queue_count = cursor.fetchone()[0] + + # Count schedule + cursor.execute("SELECT COUNT(*) FROM schedule") + schedule_count = cursor.fetchone()[0] + + conn.close() + except Exception as e: + logger.debug(f"SqliteStorage count error: {e}") + + elif storage_type in ['RedisStorage', 'RedisHuey']: + # RedisStorage: Use Redis commands + try: + if hasattr(storage, 'conn'): + # Queue is a list + queue_count = storage.conn.llen(f"{storage.name}:queue") + + # Schedule is a sorted set + schedule_count = storage.conn.zcard(f"{storage.name}:schedule") + except Exception as e: + logger.debug(f"RedisStorage count error: {e}") + + else: + # Unknown storage type - try generic attributes + try: + if hasattr(storage, 'queue_size'): + queue_count = storage.queue_size() + elif hasattr(storage, 'queue'): + queue_count = len(storage.queue) + except Exception: + pass + + try: + if hasattr(storage, 'schedule'): + schedule_count = len(storage.schedule) + except Exception: + pass + + return queue_count, schedule_count + + def get_pending_notifications_count(): """ - Get count of pending notifications in the queue (not yet processed or being retried). + Get count of pending notifications (immediate queue + scheduled/retrying). - This provides a simple count without needing to introspect individual task details, - which can vary significantly by storage backend (FileHuey, SqliteHuey, RedisHuey). + This includes: + - Tasks in the immediate queue (ready to execute now) + - Tasks in the schedule (waiting for retry or delayed execution) + + Supports FileStorage, SqliteStorage, and RedisStorage backends. Returns: Integer count of pending notifications, or None if unable to determine @@ -190,22 +282,180 @@ def get_pending_notifications_count(): return 0 try: - # Try to get queue length - # This works for most Huey storage backends - queue_length = len(huey.storage.queue) - return queue_length - except (AttributeError, TypeError): - # Some storage backends may not support len() on queue - try: - # Alternative: try to peek at queue - if hasattr(huey.storage, 'queue_size'): - return huey.storage.queue_size() - except: - pass - except Exception as e: - logger.debug(f"Unable to determine pending notification count: {e}") + # Detect storage backend type + storage_type = type(huey.storage).__name__ - return None # Unable to determine + # Get counts using backend-specific logic + queue_count, schedule_count = _count_storage_items(huey.storage, storage_type) + + total_count = queue_count + schedule_count + + if queue_count > 0: + logger.debug(f"Pending notifications - queue: {queue_count}") + if schedule_count > 0: + logger.debug(f"Pending notifications - schedule: {schedule_count}") + if total_count > 0: + logger.info(f"Total pending/retrying notifications: {total_count}") + + return total_count + + except Exception as e: + logger.error(f"Error getting pending notification count: {e}", exc_info=True) + return None # Unable to determine + + +def get_pending_notifications(limit=50): + """ + Get list of pending/retrying notifications from queue and schedule. + + Args: + limit: Maximum number to return (default: 50) + + Returns: + List of dicts with pending notification info + """ + if huey is None: + return [] + + pending = [] + import os + import pickle + import time + + try: + storage_type = type(huey.storage).__name__ + + if storage_type == 'FileStorage' and hasattr(huey.storage, 'path'): + # FileStorage: Read pickled task files + storage_path = huey.storage.path + + # Get queued tasks (immediate) + queue_dir = os.path.join(storage_path, 'queue') + if os.path.exists(queue_dir): + for root, dirs, files in os.walk(queue_dir): + for filename in files: + if filename.startswith('.') or len(pending) >= limit: + continue + filepath = os.path.join(root, filename) + try: + with open(filepath, 'rb') as f: + task_data = pickle.load(f) + notification_data = task_data.get('args', [{}])[0] if task_data.get('args') else {} + pending.append({ + 'status': 'queued', + 'watch_url': notification_data.get('watch_url', 'Unknown'), + 'watch_uuid': notification_data.get('uuid'), + 'queued_at': task_data.get('execute_time'), + }) + except Exception: + pass + + # Get scheduled tasks (retrying) + schedule_dir = os.path.join(storage_path, 'schedule') + if os.path.exists(schedule_dir): + for root, dirs, files in os.walk(schedule_dir): + for filename in files: + if filename.startswith('.') or len(pending) >= limit: + continue + filepath = os.path.join(root, filename) + try: + with open(filepath, 'rb') as f: + task_data = pickle.load(f) + notification_data = task_data.get('args', [{}])[0] if task_data.get('args') else {} + eta = task_data.get('eta') + pending.append({ + 'status': 'retrying', + 'watch_url': notification_data.get('watch_url', 'Unknown'), + 'watch_uuid': notification_data.get('uuid'), + 'retry_at': eta, + 'retry_in_seconds': int(eta - time.time()) if eta else 0, + }) + except Exception: + pass + + elif storage_type in ['SqliteStorage', 'SqliteHuey'] and hasattr(huey.storage, 'filename'): + # SqliteStorage: Query database + import sqlite3 + conn = sqlite3.connect(huey.storage.filename) + cursor = conn.cursor() + + # Get queued tasks + cursor.execute("SELECT data FROM queue LIMIT ?", (limit,)) + for row in cursor.fetchall(): + try: + task_data = pickle.loads(row[0]) + notification_data = task_data.get('args', [{}])[0] if task_data.get('args') else {} + pending.append({ + 'status': 'queued', + 'watch_url': notification_data.get('watch_url', 'Unknown'), + 'watch_uuid': notification_data.get('uuid'), + }) + except Exception: + pass + + # Get scheduled tasks + cursor.execute("SELECT data, eta FROM schedule LIMIT ?", (limit - len(pending),)) + for row in cursor.fetchall(): + try: + task_data = pickle.loads(row[0]) + notification_data = task_data.get('args', [{}])[0] if task_data.get('args') else {} + eta = row[1] + pending.append({ + 'status': 'retrying', + 'watch_url': notification_data.get('watch_url', 'Unknown'), + 'watch_uuid': notification_data.get('uuid'), + 'retry_at': eta, + 'retry_in_seconds': int(eta - time.time()) if eta else 0, + }) + except Exception: + pass + + conn.close() + + # Format timestamps for display + from changedetectionio.notification_service import timestamp_to_localtime + for item in pending: + if item.get('queued_at'): + item['queued_at_formatted'] = timestamp_to_localtime(item['queued_at']) + if item.get('retry_at'): + item['retry_at_formatted'] = timestamp_to_localtime(item['retry_at']) + + except Exception as e: + logger.error(f"Error getting pending notifications: {e}", exc_info=True) + + return pending + + +def get_last_successful_notification(): + """ + Get the most recent successful notification for reference. + + Returns: + Dict with success info or None if no successful notifications yet + """ + if huey is None or not hasattr(huey.storage, 'path'): + return None + + import os + import json + + try: + success_file = os.path.join(huey.storage.path, 'last_successful_notification.json') + if os.path.exists(success_file): + with open(success_file, 'r') as f: + success_data = json.load(f) + + # Format timestamp for display + from changedetectionio.notification_service import timestamp_to_localtime + success_time = success_data.get('timestamp') + if success_time: + success_data['timestamp_formatted'] = timestamp_to_localtime(success_time) + + return success_data + except Exception as e: + logger.debug(f"Unable to load last successful notification: {e}") + + return None def get_failed_notifications(limit=100, max_age_days=30): @@ -233,27 +483,100 @@ def get_failed_notifications(limit=100, max_age_days=30): try: # Query Huey's result storage for failed tasks - # Note: This requires accessing Huey's internal storage - from huey.storage import PeeweeStorage + # Different storage backends work differently + cutoff_time = time.time() - (max_age_days * 86400) - # Get all results and filter for errors - # Huey stores results with task IDs as keys - results = huey.storage.result_store.flush() - cutoff_time = time.time() - (max_age_days * 86400) # Convert days to seconds + results = {} + + # Try to get results - method varies by storage backend + try: + # SqliteHuey/RedisHuey have result_store.flush() + results = huey.storage.result_store.flush() + except AttributeError: + # FileStorage doesn't have result_store.flush() + # Need to enumerate result files directly from filesystem + import os + import pickle + + try: + # FileStorage stores results as pickled files in subdirectories + # Path structure: {storage.path}/results/{hash_subdir}/... + storage_path = huey.storage.path + results_dir = os.path.join(storage_path, 'results') + + if os.path.exists(results_dir): + # Walk through all subdirectories to find result files + for root, dirs, files in os.walk(results_dir): + for filename in files: + if filename.startswith('.'): + continue + + filepath = os.path.join(root, filename) + try: + # Read and unpickle the result + # Huey FileStorage format: 4-byte length + task_id + pickled data + with open(filepath, 'rb') as f: + # Read the task ID header (length-prefixed) + task_id_len_bytes = f.read(4) + if len(task_id_len_bytes) < 4: + raise EOFError("Incomplete header") + task_id_len = struct.unpack('>I', task_id_len_bytes)[0] + task_id_bytes = f.read(task_id_len) + if len(task_id_bytes) < task_id_len: + raise EOFError("Incomplete task ID") + task_id = task_id_bytes.decode('utf-8') + + # Now unpickle the result data + result_data = pickle.load(f) + results[task_id] = result_data + except (pickle.UnpicklingError, EOFError) as e: + # Corrupted or incomplete result file + # This can happen if: + # - Process crashed during write + # - Disk full + # - Leftover from interrupted shutdown + file_size = os.path.getsize(filepath) + logger.warning(f"Corrupted result file {filename} ({file_size} bytes) - likely from interrupted write. Moving to lost-found.") + try: + # Move to lost-found directory instead of deleting + import shutil + lost_found_dir = os.path.join(storage_path, 'lost-found', 'results') + os.makedirs(lost_found_dir, exist_ok=True) + + # Add timestamp to filename to avoid collisions + import time + timestamp = int(time.time()) + lost_found_path = os.path.join(lost_found_dir, f"{filename}.{timestamp}.corrupted") + + shutil.move(filepath, lost_found_path) + logger.info(f"Moved corrupted file to {lost_found_path}") + except Exception as move_err: + logger.error(f"Unable to move corrupted file to lost-found: {move_err}") + except Exception as e: + logger.debug(f"Unable to read result file {filename}: {e}") + # Note: Not logging when results_dir doesn't exist - this is normal when no failures yet + except Exception as e: + logger.debug(f"Unable to enumerate FileStorage results: {e}") + + # Import Huey's Error class for checking failed tasks + from huey.utils import Error as HueyError for task_id, result in results.items(): - if isinstance(result, Exception): - # This is a failed task - # Try to extract notification data from task args + if isinstance(result, (Exception, HueyError)): + # This is a failed task (either Exception or Huey Error object) + # Try to extract notification data from task metadata storage try: - task_data = huey.storage.get(task_id) - if task_data: - task_time = task_data.get('execute_time', 0) + # Get task metadata from our metadata storage + task_metadata = _get_task_metadata(task_id) + if task_metadata: + task_time = task_metadata.get('timestamp', 0) + notification_data = task_metadata.get('notification_data', {}) # Auto-cleanup old failed notifications to free memory if task_time and task_time < cutoff_time: logger.info(f"Auto-deleting old failed notification {task_id} (age: {(time.time() - task_time) / 86400:.1f} days)") huey.storage.delete(task_id) + _delete_task_metadata(task_id) continue # Format timestamp for display with locale awareness @@ -261,13 +584,37 @@ def get_failed_notifications(limit=100, max_age_days=30): timestamp_formatted = timestamp_to_localtime(task_time) if task_time else 'Unknown' days_ago = int((time.time() - task_time) / 86400) if task_time else 0 + # Load retry attempts for this notification (by watch_uuid) + retry_attempts = [] + notification_watch_uuid = notification_data.get('uuid') + if notification_watch_uuid and hasattr(huey.storage, 'path'): + import os + import json + import glob + + attempts_dir = os.path.join(huey.storage.path, 'retry_attempts') + if os.path.exists(attempts_dir): + attempt_pattern = os.path.join(attempts_dir, f"{notification_watch_uuid}.*.json") + for attempt_file in sorted(glob.glob(attempt_pattern)): + try: + with open(attempt_file, 'r') as f: + attempt_data = json.load(f) + # Format timestamp for display + attempt_time = attempt_data.get('timestamp') + if attempt_time: + attempt_data['timestamp_formatted'] = timestamp_to_localtime(attempt_time) + retry_attempts.append(attempt_data) + except Exception as ae: + logger.debug(f"Unable to load retry attempt file {attempt_file}: {ae}") + failed_tasks.append({ 'task_id': task_id, - 'timestamp': task_data.get('execute_time'), + 'timestamp': task_time, 'timestamp_formatted': timestamp_formatted, 'days_ago': days_ago, 'error': str(result), - 'notification_data': task_data.get('args', [{}])[0] if task_data.get('args') else {}, + 'notification_data': notification_data, + 'retry_attempts': retry_attempts, }) except Exception as e: logger.error(f"Error extracting failed task data: {e}") @@ -299,23 +646,27 @@ def retry_failed_notification(task_id): return False try: - # Get the original task data - task_data = huey.storage.get(task_id) + # Get the original task metadata from our storage + task_metadata = _get_task_metadata(task_id) - if not task_data: - logger.error(f"Task {task_id} not found in storage") + if not task_metadata: + logger.error(f"Task metadata for {task_id} not found in storage") return False - # Extract notification data and re-queue - notification_data = task_data.get('args', [{}])[0] if task_data.get('args') else {} + # Extract notification data + notification_data = task_metadata.get('notification_data', {}) if notification_data: - # Queue it again with current settings - send_notification_task(notification_data) + # Queue it again with current settings using queue_notification + # which will store new metadata for the new task + queue_notification(notification_data) # Remove from dead letter queue (it will go back if it fails again) huey.storage.delete(task_id) + # Clean up old metadata + _delete_task_metadata(task_id) + logger.info(f"Re-queued failed notification task {task_id} and removed from dead letter queue") return True else: @@ -345,13 +696,13 @@ def retry_all_failed_notifications(): failed_count = 0 try: - from huey.storage import PeeweeStorage + from huey.utils import Error as HueyError # Get all failed tasks results = huey.storage.result_store.flush() for task_id, result in results.items(): - if isinstance(result, Exception): + if isinstance(result, (Exception, HueyError)): # Try to retry this failed notification if retry_failed_notification(task_id): success_count += 1 @@ -451,8 +802,41 @@ def send_notification_task(n_object_dict): # notifications like filter failures that have custom titles and bodies. # Process and send the notification using shared datastore + # Capture Apprise logs during send + apprise_logs = [] if n_object.get('notification_urls'): - sent_obj = process_notification(n_object, datastore) + import logging + import io + + # Create a string buffer to capture Apprise logs + log_capture = io.StringIO() + handler = logging.StreamHandler(log_capture) + handler.setLevel(logging.DEBUG) + formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') + handler.setFormatter(formatter) + + # Add handler to Apprise logger + apprise_logger = logging.getLogger('apprise') + apprise_logger.addHandler(handler) + + try: + sent_obj = process_notification(n_object, datastore) + + # Capture the logs with limits to prevent excessive growth + log_output = log_capture.getvalue() + if log_output: + apprise_logs = log_output.strip().split('\n') + + # Limit: Keep only last 50 lines to prevent bloat + if len(apprise_logs) > 50: + apprise_logs = apprise_logs[-50:] + + # Limit: Truncate each line to 500 chars max + apprise_logs = [line[:500] + '...' if len(line) > 500 else line for line in apprise_logs] + finally: + # Always remove the handler + apprise_logger.removeHandler(handler) + log_capture.close() # Clear any previous error on success watch_uuid = n_object.get('uuid') @@ -468,6 +852,38 @@ def send_notification_task(n_object_dict): while len(notification_debug_log) > 100: notification_debug_log.pop(0) + # Clean up retry attempt files on success and store last successful notification + try: + import os + import glob + + if huey and hasattr(huey.storage, 'path'): + watch_uuid = n_object.get('uuid') + if watch_uuid: + attempts_dir = os.path.join(huey.storage.path, 'retry_attempts') + if os.path.exists(attempts_dir): + # Delete all attempt files for this watch + attempt_pattern = os.path.join(attempts_dir, f"{watch_uuid}.*.json") + for attempt_file in glob.glob(attempt_pattern): + os.remove(attempt_file) + logger.debug(f"Cleaned up retry attempt files for successful watch {watch_uuid}") + + # Store last successful notification for reference + # Note: This file is overwritten on each success (only keeps most recent) + # Logs are limited to 50 lines x 500 chars = ~25KB max + success_file = os.path.join(huey.storage.path, 'last_successful_notification.json') + success_data = { + 'timestamp': time.time(), + 'watch_url': n_object.get('watch_url'), + 'watch_uuid': n_object.get('uuid'), + 'notification_urls': list(n_object.get('notification_urls', {}).keys()) if n_object.get('notification_urls') else [], + 'apprise_logs': apprise_logs if apprise_logs else [], + } + with open(success_file, 'w') as f: + json.dump(success_data, f, indent=2) + except Exception as cleanup_error: + logger.debug(f"Unable to cleanup retry attempts: {cleanup_error}") + logger.success(f"Notification sent successfully for {n_object.get('watch_url')}") return sent_obj @@ -475,6 +891,44 @@ def send_notification_task(n_object_dict): # Log error and update watch with error message (preserve original error handling) logger.error(f"Watch URL: {n_object.get('watch_url')} Error {str(e)}") + # Store retry attempt details with Apprise logs + # Note: We use watch_uuid as the identifier since Huey doesn't expose task ID easily + try: + import time + import os + import uuid + + if huey and hasattr(huey.storage, 'path'): + attempts_dir = os.path.join(huey.storage.path, 'retry_attempts') + os.makedirs(attempts_dir, exist_ok=True) + + # Use watch UUID as identifier (or generate one for test notifications) + watch_uuid = n_object.get('uuid', str(uuid.uuid4())) + + # Count existing attempts for this watch + attempt_files = [f for f in os.listdir(attempts_dir) if f.startswith(f"{watch_uuid}.")] + attempt_number = len(attempt_files) + 1 + + # Store with timestamp to avoid collisions + timestamp = int(time.time()) + attempt_file = os.path.join(attempts_dir, f"{watch_uuid}.{attempt_number}.{timestamp}.json") + + attempt_data = { + 'watch_uuid': watch_uuid, + 'attempt_number': attempt_number, + 'timestamp': time.time(), + 'watch_url': n_object.get('watch_url'), + 'error': str(e), # Includes Apprise logs from exception message + 'will_retry': attempt_number <= NOTIFICATION_RETRY_COUNT + } + + with open(attempt_file, 'w') as f: + json.dump(attempt_data, f, indent=2) + + logger.debug(f"Stored retry attempt {attempt_number} for watch {watch_uuid}") + except Exception as store_error: + logger.debug(f"Unable to store retry attempt: {store_error}") + watch_uuid = n_object.get('uuid') # UUID wont be present when we submit a 'test' from the global settings @@ -508,6 +962,87 @@ def send_notification_task(n_object_dict): # Decorator will be applied after huey is initialized # This is set up in init_huey_task() +def _store_task_metadata(task_id, n_object_dict): + """Store notification metadata for a task so we can retrieve it later when task fails.""" + if not huey or not hasattr(huey.storage, 'path'): + return + + try: + import json + metadata_dir = os.path.join(huey.storage.path, 'task_metadata') + os.makedirs(metadata_dir, exist_ok=True) + + metadata_file = os.path.join(metadata_dir, f"{task_id}.json") + metadata = { + 'task_id': task_id, + 'timestamp': time.time(), + 'notification_data': n_object_dict + } + + with open(metadata_file, 'w') as f: + json.dump(metadata, f, indent=2) + except Exception as e: + logger.debug(f"Unable to store task metadata: {e}") + + +def _get_task_metadata(task_id): + """Retrieve notification metadata for a task ID.""" + if not huey or not hasattr(huey.storage, 'path'): + return None + + try: + import json + metadata_dir = os.path.join(huey.storage.path, 'task_metadata') + metadata_file = os.path.join(metadata_dir, f"{task_id}.json") + + if os.path.exists(metadata_file): + with open(metadata_file, 'r') as f: + return json.load(f) + except Exception as e: + logger.debug(f"Unable to load task metadata for {task_id}: {e}") + + return None + + +def _delete_task_metadata(task_id): + """Delete task metadata file (cleanup after success or manual deletion).""" + if not huey or not hasattr(huey.storage, 'path'): + return + + try: + metadata_dir = os.path.join(huey.storage.path, 'task_metadata') + metadata_file = os.path.join(metadata_dir, f"{task_id}.json") + + if os.path.exists(metadata_file): + os.remove(metadata_file) + except Exception as e: + logger.debug(f"Unable to delete task metadata for {task_id}: {e}") + + +def queue_notification(n_object_dict): + """ + Queue a notification task and store its metadata for later retrieval. + + This is the main entry point for queueing notifications. It wraps + send_notification_task() and stores the task metadata so we can + retrieve notification details even after the task completes. + + Args: + n_object_dict: Notification data dictionary + + Returns: + Huey TaskResultWrapper with task ID + """ + # Queue the task with Huey + task_result = send_notification_task(n_object_dict) + + # Store metadata so we can retrieve it later + if task_result and hasattr(task_result, 'id'): + _store_task_metadata(task_result.id, n_object_dict) + + return task_result + + def init_huey_task(): """ Decorate send_notification_task with Huey task decorator. @@ -519,21 +1054,145 @@ def init_huey_task(): raise RuntimeError("Huey not initialized! Call init_huey(datastore_path) first") # Apply Huey task decorator with exponential backoff retry settings - retry_delays = get_retry_delays() + # backoff=2 means each retry delay is 2x the previous (exponential backoff) send_notification_task = huey.task( retries=NOTIFICATION_RETRY_COUNT, - retry_delay=retry_delays if retry_delays else NOTIFICATION_RETRY_DELAY + retry_delay=NOTIFICATION_RETRY_DELAY, + backoff=2 # Exponential backoff multiplier (60s → 120s → 240s → ...) )(send_notification_task) + retry_delays = get_retry_delays() if retry_delays: - logger.info(f"Notification retry configuration: {NOTIFICATION_RETRY_COUNT} retries with exponential backoff: {retry_delays}") + logger.info(f"Notification retry configuration: {NOTIFICATION_RETRY_COUNT} retries with exponential backoff (base: {NOTIFICATION_RETRY_DELAY}s, delays: {retry_delays})") else: logger.info(f"Notification retry configuration: No retries configured") +def clear_all_notifications(): + """ + Clear ALL notifications from queue, schedule, results, and retry attempts. + + WARNING: This is a destructive operation that clears: + - Immediate queue (pending notifications) + - Schedule (retrying/delayed notifications) + - Results (failed notifications) + - Retry attempt files + + Returns: + Dict with counts of cleared items + """ + if huey is None: + return {'error': 'Huey not initialized'} + + import os + import shutil + + cleared = { + 'queue': 0, + 'schedule': 0, + 'results': 0, + 'retry_attempts': 0, + 'task_metadata': 0 + } + + try: + storage_type = type(huey.storage).__name__ + + if storage_type == 'FileStorage' and hasattr(huey.storage, 'path'): + # FileStorage: Delete directory contents + storage_path = huey.storage.path + + # Clear queue + queue_dir = os.path.join(storage_path, 'queue') + if os.path.exists(queue_dir): + for root, dirs, files in os.walk(queue_dir): + for f in files: + if not f.startswith('.'): + os.remove(os.path.join(root, f)) + cleared['queue'] += 1 + + # Clear schedule + schedule_dir = os.path.join(storage_path, 'schedule') + if os.path.exists(schedule_dir): + for root, dirs, files in os.walk(schedule_dir): + for f in files: + if not f.startswith('.'): + os.remove(os.path.join(root, f)) + cleared['schedule'] += 1 + + # Clear results + results_dir = os.path.join(storage_path, 'results') + if os.path.exists(results_dir): + for root, dirs, files in os.walk(results_dir): + for f in files: + if not f.startswith('.'): + os.remove(os.path.join(root, f)) + cleared['results'] += 1 + + # Clear retry attempts + attempts_dir = os.path.join(storage_path, 'retry_attempts') + if os.path.exists(attempts_dir): + for f in os.listdir(attempts_dir): + if f.endswith('.json'): + os.remove(os.path.join(attempts_dir, f)) + cleared['retry_attempts'] += 1 + + # Clear task metadata + metadata_dir = os.path.join(storage_path, 'task_metadata') + if os.path.exists(metadata_dir): + for f in os.listdir(metadata_dir): + if f.endswith('.json'): + os.remove(os.path.join(metadata_dir, f)) + cleared['task_metadata'] += 1 + + elif storage_type in ['SqliteStorage', 'SqliteHuey'] and hasattr(huey.storage, 'filename'): + # SqliteStorage: Delete from tables + import sqlite3 + conn = sqlite3.connect(huey.storage.filename) + cursor = conn.cursor() + + cursor.execute("DELETE FROM queue") + cleared['queue'] = cursor.rowcount + + cursor.execute("DELETE FROM schedule") + cleared['schedule'] = cursor.rowcount + + cursor.execute("DELETE FROM results") + cleared['results'] = cursor.rowcount + + conn.commit() + conn.close() + + elif storage_type in ['RedisStorage', 'RedisHuey'] and hasattr(huey.storage, 'conn'): + # RedisStorage: Delete keys + name = huey.storage.name + + # Clear queue (list) + cleared['queue'] = huey.storage.conn.llen(f"{name}:queue") + huey.storage.conn.delete(f"{name}:queue") + + # Clear schedule (sorted set) + cleared['schedule'] = huey.storage.conn.zcard(f"{name}:schedule") + huey.storage.conn.delete(f"{name}:schedule") + + # Clear results (hash or keys) + # Note: This depends on how Huey stores results in Redis + result_keys = huey.storage.conn.keys(f"{name}:result:*") + if result_keys: + cleared['results'] = len(result_keys) + huey.storage.conn.delete(*result_keys) + + logger.warning(f"Cleared all notifications: {cleared}") + return cleared + + except Exception as e: + logger.error(f"Error clearing notifications: {e}", exc_info=True) + return {'error': str(e)} + + def cleanup_old_failed_notifications(max_age_days=30): """ - Clean up failed notifications older than max_age_days. + Clean up failed notifications and retry attempts older than max_age_days. Called on startup to prevent indefinite accumulation of old failures. @@ -547,29 +1206,42 @@ def cleanup_old_failed_notifications(max_age_days=30): return 0 import time + import os deleted_count = 0 try: - results = huey.storage.result_store.flush() + # Use get_failed_notifications with auto-cleanup to handle this + # It already has logic to delete old failed notifications + # We just call it and let it do the cleanup cutoff_time = time.time() - (max_age_days * 86400) - for task_id, result in results.items(): - if isinstance(result, Exception): - try: - task_data = huey.storage.get(task_id) - if task_data: - task_time = task_data.get('execute_time', 0) - if task_time and task_time < cutoff_time: - huey.storage.delete(task_id) - deleted_count += 1 - except Exception as e: - logger.error(f"Error cleaning up old failed notification {task_id}: {e}") + # FileStorage and other backends handle result storage differently + # The get_failed_notifications function already handles cleanup + # So we just trigger it here + get_failed_notifications(limit=1000, max_age_days=max_age_days) - if deleted_count > 0: - logger.info(f"Cleaned up {deleted_count} old failed notifications (older than {max_age_days} days)") + # Also clean up old retry attempt files + if hasattr(huey.storage, 'path'): + attempts_dir = os.path.join(huey.storage.path, 'retry_attempts') + if os.path.exists(attempts_dir): + for filename in os.listdir(attempts_dir): + if filename.endswith('.json'): + filepath = os.path.join(attempts_dir, filename) + try: + file_mtime = os.path.getmtime(filepath) + if file_mtime < cutoff_time: + os.remove(filepath) + deleted_count += 1 + except Exception as fe: + logger.debug(f"Unable to delete old retry attempt file {filename}: {fe}") + + if deleted_count > 0: + logger.info(f"Cleaned up {deleted_count} old retry attempt files (older than {max_age_days} days)") + + logger.info(f"Completed cleanup check for failed notifications older than {max_age_days} days") except Exception as e: - logger.error(f"Error during failed notification cleanup: {e}") + logger.debug(f"Unable to cleanup old failed notifications: {e}") return deleted_count