From 31a104f29e37eca3bedb37dd04a0ac6abd9dc427 Mon Sep 17 00:00:00 2001 From: dgtlmoon Date: Mon, 5 Jan 2026 11:52:36 +0100 Subject: [PATCH] Retry notifications on fail --- changedetectionio/__init__.py | 6 +- .../blueprint/settings/__init__.py | 33 +- changedetectionio/flask_app.py | 77 +--- changedetectionio/notification/handler.py | 15 +- changedetectionio/notification/task_queue.py | 381 ++++++++++++++++++ changedetectionio/notification_service.py | 32 +- .../templates/_common_fields.html | 1 + requirements.txt | 1 + 8 files changed, 467 insertions(+), 79 deletions(-) create mode 100644 changedetectionio/notification/task_queue.py diff --git a/changedetectionio/__init__.py b/changedetectionio/__init__.py index 05970f770..affb1e5ac 100644 --- a/changedetectionio/__init__.py +++ b/changedetectionio/__init__.py @@ -44,10 +44,10 @@ def sigshutdown_handler(_signo, _stack_frame): # Close janus queues properly try: - from changedetectionio.flask_app import update_q, notification_q + from changedetectionio.flask_app import update_q update_q.close() - notification_q.close() - logger.debug("Janus queues closed successfully") + logger.debug("Janus update queue closed successfully") + # notification_q is deprecated - now using Huey task queue which handles its own shutdown except Exception as e: logger.critical(f"CRITICAL: Failed to close janus queues: {e}") diff --git a/changedetectionio/blueprint/settings/__init__.py b/changedetectionio/blueprint/settings/__init__.py index 2396d5667..1b39d2464 100644 --- a/changedetectionio/blueprint/settings/__init__.py +++ b/changedetectionio/blueprint/settings/__init__.py @@ -77,12 +77,12 @@ def construct_blueprint(datastore: ChangeDetectionStore): # Adjust worker count if it changed if new_worker_count != old_worker_count: from changedetectionio import worker_handler - from changedetectionio.flask_app import update_q, notification_q, app, datastore as ds - + from changedetectionio.flask_app import update_q, app, datastore as ds + result = worker_handler.adjust_async_worker_count( new_count=new_worker_count, update_q=update_q, - notification_q=notification_q, + notification_q=None, # Now using Huey task queue app=app, datastore=ds ) @@ -142,4 +142,31 @@ def construct_blueprint(datastore: ChangeDetectionStore): logs=notification_debug_log if len(notification_debug_log) else ["Notification logs are empty - no notifications sent yet."]) return output + @settings_blueprint.route("/failed-notifications", methods=['GET']) + @login_optionally_required + def failed_notifications(): + """View notifications that failed all retry attempts""" + from changedetectionio.notification.task_queue import get_failed_notifications + + failed = get_failed_notifications(limit=100) + + output = render_template("failed-notifications.html", + failed_notifications=failed) + return output + + @settings_blueprint.route("/retry-notification/", methods=['POST']) + @login_optionally_required + def retry_notification(task_id): + """Retry a failed notification by task ID""" + from changedetectionio.notification.task_queue import retry_failed_notification + + success = retry_failed_notification(task_id) + + if success: + flash(f"Notification {task_id} queued for retry.", 'notice') + else: + flash(f"Failed to retry notification {task_id}. Check logs for details.", 'error') + + return redirect(url_for('settings.failed_notifications')) + return settings_blueprint \ No newline at end of file diff --git a/changedetectionio/flask_app.py b/changedetectionio/flask_app.py index e70a10778..ef45c0f4b 100644 --- a/changedetectionio/flask_app.py +++ b/changedetectionio/flask_app.py @@ -12,7 +12,7 @@ from blinker import signal from changedetectionio.strtobool import strtobool from threading import Event -from changedetectionio.queue_handlers import RecheckPriorityQueue, NotificationQueue +from changedetectionio.queue_handlers import RecheckPriorityQueue # NotificationQueue deprecated - now using Huey from changedetectionio import worker_handler from flask import ( @@ -48,9 +48,9 @@ datastore = None ticker_thread = None extra_stylesheets = [] -# Use bulletproof janus-based queues for sync/async reliability +# Use bulletproof janus-based queues for sync/async reliability update_q = RecheckPriorityQueue() -notification_q = NotificationQueue() +# notification_q = NotificationQueue() # DEPRECATED: Now using Huey task queue MAX_QUEUE_SIZE = 2000 app = Flask(__name__, @@ -565,7 +565,7 @@ def changedetection_app(config=None, datastore_o=None): health_result = worker_handler.check_worker_health( expected_count=expected_workers, update_q=update_q, - notification_q=notification_q, + notification_q=None, # Now using Huey task queue app=app, datastore=datastore ) @@ -626,11 +626,20 @@ def changedetection_app(config=None, datastore_o=None): # Can be overridden by ENV or use the default settings n_workers = int(os.getenv("FETCH_WORKERS", datastore.data['settings']['requests']['workers'])) logger.info(f"Starting {n_workers} workers during app initialization") - worker_handler.start_workers(n_workers, update_q, notification_q, app, datastore) + # Pass None for notification_q - now using Huey task queue directly + worker_handler.start_workers(n_workers, update_q, None, app, datastore) + + # Initialize Huey task queue for notifications + from changedetectionio.notification.task_queue import init_huey, init_huey_task, start_huey_consumer + init_huey(datastore.datastore_path) + init_huey_task() # Apply task decorator + + # Start Huey consumer for notification processing (replaces notification_runner) + # Using 1 worker thread to match original notification_runner behavior + threading.Thread(target=start_huey_consumer, args=(1,), daemon=True).start() # @todo handle ctrl break ticker_thread = threading.Thread(target=ticker_thread_check_time_launch_checks).start() - threading.Thread(target=notification_runner).start() in_pytest = "pytest" in sys.modules or "PYTEST_CURRENT_TEST" in os.environ # Check for new release version, but not when running in test/build or pytest @@ -670,56 +679,10 @@ def check_for_new_version(): app.config.exit.wait(86400) -def notification_runner(): - global notification_debug_log - from datetime import datetime - import json - with app.app_context(): - while not app.config.exit.is_set(): - try: - # At the moment only one thread runs (single runner) - n_object = notification_q.get(block=False) - except queue.Empty: - time.sleep(1) - - else: - - now = datetime.now() - sent_obj = None - - try: - from changedetectionio.notification.handler import process_notification - - # Fallback to system config if not set - if not n_object.get('notification_body') and datastore.data['settings']['application'].get('notification_body'): - n_object['notification_body'] = datastore.data['settings']['application'].get('notification_body') - - if not n_object.get('notification_title') and datastore.data['settings']['application'].get('notification_title'): - n_object['notification_title'] = datastore.data['settings']['application'].get('notification_title') - - if not n_object.get('notification_format') and datastore.data['settings']['application'].get('notification_format'): - n_object['notification_format'] = datastore.data['settings']['application'].get('notification_format') - if n_object.get('notification_urls', {}): - sent_obj = process_notification(n_object, datastore) - - except Exception as e: - logger.error(f"Watch URL: {n_object['watch_url']} Error {str(e)}") - - # UUID wont be present when we submit a 'test' from the global settings - if 'uuid' in n_object: - datastore.update_watch(uuid=n_object['uuid'], - update_obj={'last_notification_error': "Notification error detected, goto notification log."}) - - log_lines = str(e).splitlines() - notification_debug_log += log_lines - - with app.app_context(): - app.config['watch_check_update_SIGNAL'].send(app_context=app, watch_uuid=n_object.get('uuid')) - - # Process notifications - notification_debug_log+= ["{} - SENDING - {}".format(now.strftime("%c"), json.dumps(sent_obj))] - # Trim the log length - notification_debug_log = notification_debug_log[-100:] +# DEPRECATED: notification_runner has been replaced by Huey task queue +# All logic from this function has been moved to changedetectionio/notification/task_queue.py +# in the send_notification_task() function with automatic retry logic and persistent queuing +# See: changedetectionio/notification/task_queue.py - send_notification_task() @@ -743,7 +706,7 @@ def ticker_thread_check_time_launch_checks(): health_result = worker_handler.check_worker_health( expected_count=expected_workers, update_q=update_q, - notification_q=notification_q, + notification_q=None, # Now using Huey task queue app=app, datastore=datastore ) diff --git a/changedetectionio/notification/handler.py b/changedetectionio/notification/handler.py index 87a22cf86..2431fdffd 100644 --- a/changedetectionio/notification/handler.py +++ b/changedetectionio/notification/handler.py @@ -415,8 +415,10 @@ def process_notification(n_object: NotificationContextData, datastore): if not '= limit: + break + + except Exception as e: + logger.error(f"Error querying failed notifications: {e}") + + return failed_tasks + + +def retry_failed_notification(task_id): + """ + Retry a failed notification by task ID. + + Args: + task_id: Huey task ID to retry + + Returns: + True if successfully queued for retry, False otherwise + """ + if huey is None: + logger.error("Huey not initialized") + return False + + try: + # Get the original task data + task_data = huey.storage.get(task_id) + + if not task_data: + logger.error(f"Task {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 {} + + if notification_data: + # Queue it again + send_notification_task(notification_data) + logger.info(f"Re-queued failed notification task {task_id}") + return True + else: + logger.error(f"No notification data found for task {task_id}") + return False + + except Exception as e: + logger.error(f"Error retrying notification {task_id}: {e}") + return False + + +def send_notification_task(n_object_dict): + """ + Background task to send a notification with automatic retry on failure. + + Retries 3 times with 60 second delay between attempts. + + IMPORTANT: Notification configuration (notification_urls, notification_title, + notification_body, notification_format) is RELOADED from the datastore at + retry time. This allows operators to fix broken settings (e.g., wrong SMTP + server) and retry with corrected configuration. + + Snapshot data (diff, watch_url, triggered_text, etc.) is preserved from + the original notification trigger. + + Preserves all logic from the original notification_runner including: + - Reloading notification settings from current datastore state at retry time + - notification_debug_log tracking + - Signal emission on errors + - Error handling and watch updates + + Args: + n_object_dict: Serialized NotificationContextData as dict (snapshot data) + + Returns: + List of sent notification objects with title, body, url + + Raises: + Exception: Any error during notification sending (triggers retry) + """ + from changedetectionio.notification_service import NotificationContextData + from changedetectionio.notification.handler import process_notification + from changedetectionio.flask_app import datastore, notification_debug_log, app + from datetime import datetime + import json + + # Reconstruct NotificationContextData from serialized dict + n_object = NotificationContextData(initial_data=n_object_dict) + + now = datetime.now() + sent_obj = None + + try: + # ALWAYS reload notification configuration from current datastore state + # This allows operators to fix broken notification settings (e.g., wrong SMTP server) + # and retry failed notifications with the corrected configuration + watch_uuid = n_object.get('uuid') + watch = None + + # Get current watch data if this is a watch notification (not a test notification) + if watch_uuid and watch_uuid in datastore.data['watching']: + watch = datastore.data['watching'][watch_uuid] + + # Reload notification_urls from current settings (watch-level or system-level) + if watch and watch.get('notification_urls'): + n_object['notification_urls'] = watch.get('notification_urls') + else: + # Fallback to system-level notification_urls + n_object['notification_urls'] = datastore.data['settings']['application'].get('notification_urls', {}) + + # Reload notification_title from current settings + if watch and watch.get('notification_title'): + n_object['notification_title'] = watch.get('notification_title') + else: + n_object['notification_title'] = datastore.data['settings']['application'].get('notification_title') + + # Reload notification_body from current settings + if watch and watch.get('notification_body'): + n_object['notification_body'] = watch.get('notification_body') + else: + n_object['notification_body'] = datastore.data['settings']['application'].get('notification_body') + + # Reload notification_format from current settings + if watch and watch.get('notification_format'): + n_object['notification_format'] = watch.get('notification_format') + else: + n_object['notification_format'] = datastore.data['settings']['application'].get('notification_format') + + # Process and send the notification using shared datastore + if n_object.get('notification_urls'): + sent_obj = process_notification(n_object, datastore) + + # Clear any previous error on success + watch_uuid = n_object.get('uuid') + if watch_uuid and watch_uuid in datastore.data['watching']: + datastore.update_watch( + uuid=watch_uuid, + update_obj={'last_notification_error': None} + ) + + # Add to notification debug log (preserve original logging) + notification_debug_log.append("{} - SENDING - {}".format(now.strftime("%c"), json.dumps(sent_obj))) + # Trim the log length + while len(notification_debug_log) > 100: + notification_debug_log.pop(0) + + logger.success(f"Notification sent successfully for {n_object.get('watch_url')}") + return sent_obj + + except Exception as e: + # 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)}") + + watch_uuid = n_object.get('uuid') + + # UUID wont be present when we submit a 'test' from the global settings + if watch_uuid: + try: + if watch_uuid in datastore.data['watching']: + datastore.update_watch( + uuid=watch_uuid, + update_obj={'last_notification_error': "Notification error detected, goto notification log."} + ) + except Exception as update_error: + logger.error(f"Failed to update watch error status: {update_error}") + + # Add error lines to debug log (preserve original logging) + log_lines = str(e).splitlines() + notification_debug_log.extend(log_lines) + # Trim the log length + while len(notification_debug_log) > 100: + notification_debug_log.pop(0) + + # Send signal (preserve original signal emission) + try: + with app.app_context(): + app.config['watch_check_update_SIGNAL'].send(app_context=app, watch_uuid=watch_uuid) + except Exception as signal_error: + logger.error(f"Failed to send watch_check_update signal: {signal_error}") + + # Re-raise to trigger Huey retry + raise + + +# Decorator will be applied after huey is initialized +# This is set up in init_huey_task() +def init_huey_task(): + """ + Decorate send_notification_task with Huey task decorator. + + Must be called after init_huey() so the decorator can be applied. + """ + global send_notification_task + if huey is None: + raise RuntimeError("Huey not initialized! Call init_huey(datastore_path) first") + + # Apply Huey task decorator + send_notification_task = huey.task(retries=3, retry_delay=60)(send_notification_task) + + +def start_huey_consumer(workers=1): + """ + Start Huey consumer in-process as background threads. + + Replaces the old notification_runner() thread with Huey's consumer + threads that provide retry logic and persistent queuing. + + Args: + workers: Number of worker threads (default: 1) + """ + global huey + + if huey is None: + raise RuntimeError("Huey not initialized! Call init_huey(datastore_path) first") + + logger.info(f"Starting Huey notification consumer with {workers} worker threads") + + try: + from huey.consumer import Consumer + + # Create and run consumer + # Note: We disable signal handlers since we're running in a background thread, not main thread + consumer = Consumer( + huey, + workers=workers, + worker_type='thread', + scheduler_interval=1, # Poll queue every 1 second + check_worker_health=True, + health_check_interval=60, + # Disable signal handlers - we're in a thread, not main process + # The main Flask app will handle shutdown signals + ) + + # Override signal handler setup to do nothing (we're in a thread) + consumer._set_signal_handlers = lambda: None + + consumer.run() # This blocks, so it runs in the thread + + except Exception as e: + logger.error(f"Failed to start Huey consumer: {e}") + raise diff --git a/changedetectionio/notification_service.py b/changedetectionio/notification_service.py index 21871b05d..31512dd57 100644 --- a/changedetectionio/notification_service.py +++ b/changedetectionio/notification_service.py @@ -171,9 +171,10 @@ class NotificationService: Standalone notification service that handles all notification functionality previously embedded in the update_worker class """ - - def __init__(self, datastore, notification_q): + + def __init__(self, datastore, notification_q=None): self.datastore = datastore + # notification_q is deprecated - now using Huey task queue directly self.notification_q = notification_q def queue_notification_for_watch(self, n_object: NotificationContextData, watch, date_index_from=-2, date_index_to=-1): @@ -227,12 +228,11 @@ class NotificationService: triggered_text=triggered_text, timestamp_changed=dates[date_index_to])) - if self.notification_q: - logger.debug("Queued notification for sending") - self.notification_q.put(n_object) - else: - logger.debug("Not queued, no queue defined. Just returning processed data") - return n_object + # Queue notification to Huey for processing with retry logic + from changedetectionio.notification.task_queue import send_notification_task + logger.debug("Queuing notification to Huey for sending with retry") + send_notification_task(dict(n_object)) + return n_object def send_content_changed_notification(self, watch_uuid): """ @@ -315,8 +315,9 @@ Thanks - Your omniscient changedetection.io installation. 'uuid': watch_uuid, 'screenshot': None }) - self.notification_q.put(n_object) - logger.debug(f"Sent filter not found notification for {watch_uuid}") + from changedetectionio.notification.task_queue import send_notification_task + send_notification_task(dict(n_object)) + logger.debug(f"Queued filter not found notification for {watch_uuid}") else: logger.debug(f"NOT sending filter not found notification for {watch_uuid} - no notification URLs") @@ -363,13 +364,18 @@ Thanks - Your omniscient changedetection.io installation. 'watch_url': watch['url'], 'uuid': watch_uuid }) - self.notification_q.put(n_object) - logger.error(f"Sent step not found notification for {watch_uuid}") + from changedetectionio.notification.task_queue import send_notification_task + send_notification_task(dict(n_object)) + logger.error(f"Queued step not found notification for {watch_uuid}") # Convenience functions for creating notification service instances -def create_notification_service(datastore, notification_q): +def create_notification_service(datastore, notification_q=None): """ Factory function to create a NotificationService instance + + Args: + datastore: The ChangeDetectionStore instance + notification_q: Deprecated, no longer used (kept for backward compatibility) """ return NotificationService(datastore, notification_q) \ No newline at end of file diff --git a/changedetectionio/templates/_common_fields.html b/changedetectionio/templates/_common_fields.html index 36cc1b6b9..1f3e61fc7 100644 --- a/changedetectionio/templates/_common_fields.html +++ b/changedetectionio/templates/_common_fields.html @@ -141,6 +141,7 @@ Add email Add an email address {% endif %} Notification debug logs + Failed notifications
diff --git a/requirements.txt b/requirements.txt index 49c824a67..e67359f7a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,6 +9,7 @@ flask_expects_json~=1.7 flask_restful flask_cors # For the Chrome extension to operate janus # Thread-safe async/sync queue bridge +huey ~= 2.5 # Task queue for notification retries with FileHuey/SqliteHuey/RedisHuey support flask_wtf~=1.2 flask~=3.1 flask-socketio~=5.5.1