mirror of
https://github.com/dgtlmoon/changedetection.io.git
synced 2026-09-15 09:57:04 +00:00
retry improvements
This commit is contained in:
dgtlmoon
@@ -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/<task_id>", 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():
|
||||
|
||||
@@ -0,0 +1,260 @@
|
||||
{% extends 'base.html' %}
|
||||
|
||||
{% block content %}
|
||||
<div class="edit-form">
|
||||
<div class="inner">
|
||||
|
||||
<h4 style="margin-top: 0px;">Failed Notifications (Exhausted Retries)</h4>
|
||||
|
||||
<!-- Last Successful Notification Reference -->
|
||||
{% if last_success %}
|
||||
<div style="background: #d4edda; border: 1px solid #c3e6cb; border-radius: 5px; padding: 15px; margin-bottom: 20px;">
|
||||
<h5 style="margin-top: 0; color: #155724;">✅ Most Recent Successful Notification</h5>
|
||||
<div style="font-size: 90%;">
|
||||
<div style="margin-bottom: 5px;">
|
||||
<strong>Timestamp:</strong> {{ last_success.timestamp_formatted }}
|
||||
</div>
|
||||
{% if last_success.watch_url %}
|
||||
<div style="margin-bottom: 5px;">
|
||||
<strong>Watch URL:</strong> <a href="{{ last_success.watch_url }}" target="_blank" style="word-break: break-all;">{{ last_success.watch_url }}</a>
|
||||
</div>
|
||||
{% endif %}
|
||||
{% if last_success.notification_urls %}
|
||||
<div style="margin-bottom: 5px;">
|
||||
<strong>Sent via:</strong>
|
||||
{% for url in last_success.notification_urls %}
|
||||
<code style="background: #fff; padding: 2px 6px; border-radius: 3px; margin-right: 5px; font-size: 85%;">{{ url }}</code>
|
||||
{% endfor %}
|
||||
</div>
|
||||
{% endif %}
|
||||
{% if last_success.apprise_logs %}
|
||||
<details style="margin-top: 10px;">
|
||||
<summary style="cursor: pointer; font-weight: bold; font-size: 90%; color: #155724;">📋 View Apprise Logs</summary>
|
||||
<pre style="background: #fff; padding: 10px; border-radius: 3px; border: 1px solid #c3e6cb; margin: 10px 0 0 0; white-space: pre-wrap; word-wrap: break-word; font-size: 80%; color: #333; max-height: 300px; overflow-y: auto;">{% for log_line in last_success.apprise_logs %}{{ log_line }}
|
||||
{% endfor %}</pre>
|
||||
</details>
|
||||
{% endif %}
|
||||
</div>
|
||||
<p style="font-size: 85%; color: #155724; margin: 10px 0 0 0; font-style: italic;">
|
||||
Use this as reference - this notification was sent successfully with current settings.
|
||||
</p>
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
<!-- Notification Queue Status Summary -->
|
||||
<div style="background: #f0f7ff; border: 1px solid #b8daff; border-radius: 5px; padding: 15px; margin-bottom: 20px;">
|
||||
<h5 style="margin-top: 0; margin-bottom: 10px;">Notification Queue Status</h5>
|
||||
<div style="display: flex; gap: 20px; flex-wrap: wrap;">
|
||||
<div style="flex: 1; min-width: 200px;">
|
||||
{% if pending_count is not none %}
|
||||
<div style="font-size: 90%;">
|
||||
<strong>🔄 Pending/Retrying:</strong>
|
||||
<span style="font-size: 120%; font-weight: bold; color: #0066cc;">{{ pending_count }}</span>
|
||||
<span style="color: #666; font-size: 85%;">notification{{ 's' if pending_count != 1 else '' }}</span>
|
||||
</div>
|
||||
<p style="font-size: 85%; color: #666; margin: 5px 0 0 0;">
|
||||
Currently in queue or being retried
|
||||
</p>
|
||||
{% else %}
|
||||
<div style="font-size: 90%; color: #666;">
|
||||
<strong>🔄 Pending/Retrying:</strong> <em>Unable to determine</em>
|
||||
</div>
|
||||
{% endif %}
|
||||
</div>
|
||||
<div style="flex: 1; min-width: 200px;">
|
||||
<div style="font-size: 90%;">
|
||||
<strong>💀 Failed (Dead Letter):</strong>
|
||||
<span style="font-size: 120%; font-weight: bold; color: {% if failed_notifications|length == 0 %}#28a745{% else %}#dc3545{% endif %};">{{ failed_notifications|length }}</span>
|
||||
<span style="color: #666; font-size: 85%;">notification{{ 's' if failed_notifications|length != 1 else '' }}</span>
|
||||
</div>
|
||||
<p style="font-size: 85%; color: #666; margin: 5px 0 0 0;">
|
||||
Exhausted all retry attempts
|
||||
</p>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- List of pending notifications -->
|
||||
{% if pending_list %}
|
||||
<details style="margin-top: 15px;">
|
||||
<summary style="cursor: pointer; font-weight: bold; font-size: 90%; color: #0066cc;">📋 View Pending/Retrying Notifications ({{ pending_list|length }})</summary>
|
||||
<div style="margin-top: 10px;">
|
||||
{% for item in pending_list %}
|
||||
<div style="background: #fff; border: 1px solid #dee2e6; border-radius: 3px; padding: 8px; margin-bottom: 5px; font-size: 85%;">
|
||||
<div style="margin-bottom: 3px;">
|
||||
<strong>{% if item.status == 'queued' %}⏳ Queued{% else %}🔄 Retrying{% endif %}:</strong>
|
||||
{% if item.watch_url %}
|
||||
<a href="{{ item.watch_url }}" target="_blank" style="word-break: break-all;">{{ item.watch_url }}</a>
|
||||
{% else %}
|
||||
<span style="color: #666;">Test notification</span>
|
||||
{% endif %}
|
||||
</div>
|
||||
{% if item.status == 'retrying' and item.retry_at_formatted %}
|
||||
<div style="color: #666; font-size: 90%;">
|
||||
Retry at: {{ item.retry_at_formatted }}
|
||||
{% if item.retry_in_seconds > 0 %}
|
||||
(in {{ item.retry_in_seconds }}s)
|
||||
{% endif %}
|
||||
</div>
|
||||
{% endif %}
|
||||
</div>
|
||||
{% endfor %}
|
||||
</div>
|
||||
</details>
|
||||
{% endif %}
|
||||
</div>
|
||||
|
||||
<!-- Retry Schedule Information -->
|
||||
<div style="background: #f8f9fa; border: 1px solid #dee2e6; border-radius: 5px; padding: 15px; margin-bottom: 20px;">
|
||||
<h5 style="margin-top: 0;">Automatic Retry Schedule (Exponential Backoff)</h5>
|
||||
<p style="font-size: 90%; margin-bottom: 10px;">
|
||||
Notifications are automatically retried <strong>{{ retry_config.retry_count }} times</strong> with <strong>exponential backoff</strong> starting at {{ retry_config.retry_delay_seconds }} seconds.
|
||||
</p>
|
||||
<table class="pure-table" style="width: 100%; font-size: 90%;">
|
||||
<thead>
|
||||
<tr>
|
||||
<th>Attempt</th>
|
||||
<th>Time</th>
|
||||
<th>Action</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
<tr>
|
||||
<td><strong>1st</strong> (initial)</td>
|
||||
<td>T+0:00</td>
|
||||
<td>⚠️ Fails (e.g., SMTP server down)</td>
|
||||
</tr>
|
||||
{% 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 %}
|
||||
<tr>
|
||||
<td><strong>{{ i + 2 }}{{ ['st', 'nd', 'rd'][i + 1] if i + 1 < 3 else 'th' }}</strong> (retry {{ i + 1 }})</td>
|
||||
<td>T+{{ '%d:%02d' % (cumulative_time // 60, cumulative_time % 60) }}</td>
|
||||
<td>{% if i < retry_config.retry_count - 1 %}⚠️ Fails → Wait {{ delay }}s ({{ '%d:%02d' % (delay // 60, delay % 60) }}){% else %}⚠️ Fails → Give up{% endif %}</td>
|
||||
</tr>
|
||||
{% endfor %}
|
||||
<tr style="background: #fff3cd;">
|
||||
<td><strong>Dead Letter</strong></td>
|
||||
<td>T+{{ '%d:%02d' % (retry_config.total_time_seconds // 60, retry_config.total_time_seconds % 60) }}</td>
|
||||
<td>💀 Moved to this failed notifications list</td>
|
||||
</tr>
|
||||
</tbody>
|
||||
</table>
|
||||
<p style="font-size: 85%; margin-top: 10px; margin-bottom: 0; color: #666;">
|
||||
<strong>Total:</strong> {{ 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.
|
||||
</p>
|
||||
</div>
|
||||
|
||||
<!-- Action Buttons (always visible) -->
|
||||
<div style="margin-bottom: 15px;">
|
||||
{% if failed_notifications|length > 0 %}
|
||||
<p style="font-size: 90%; color: #666; margin-bottom: 10px;">
|
||||
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.
|
||||
</p>
|
||||
<form method="POST" action="{{ url_for('settings.retry_all_notifications') }}" style="display: inline-block; margin-right: 10px;">
|
||||
<button type="submit" class="pure-button pure-button-primary" style="background: #28a745;" onclick="return confirm('Retry all {{ failed_notifications|length }} failed notifications?');">
|
||||
<span class="icon-repeat"></span> Retry All ({{ failed_notifications|length }})
|
||||
</button>
|
||||
</form>
|
||||
{% endif %}
|
||||
|
||||
<!-- Clear All button - always visible -->
|
||||
{% if pending_count > 0 or failed_notifications|length > 0 %}
|
||||
<form method="POST" action="{{ url_for('settings.clear_all_notifications') }}" style="display: inline-block;">
|
||||
<button type="submit" class="pure-button" style="background: #dc3545; color: white;" onclick="return confirm('⚠️ WARNING: This will DELETE ALL notifications:\n\n- {{ pending_count if pending_count else 0 }} Pending/Retrying\n- {{ failed_notifications|length }} Failed\n- All retry attempts\n\nThis action cannot be undone!\n\nAre you sure?');">
|
||||
<span class="icon-trash"></span> Clear All
|
||||
</button>
|
||||
</form>
|
||||
{% endif %}
|
||||
|
||||
{% if failed_notifications|length > 0 or pending_count > 0 %}
|
||||
<div style="font-size: 85%; color: #666; margin-top: 10px;">
|
||||
{% if failed_notifications|length > 0 %}
|
||||
<strong>Retry All:</strong> Re-queue failed notifications with current settings.<br>
|
||||
{% endif %}
|
||||
<strong>Clear All:</strong> Delete ALL pending, retrying, and failed notifications (cannot be undone).
|
||||
</div>
|
||||
{% endif %}
|
||||
</div>
|
||||
|
||||
{% if failed_notifications|length == 0 %}
|
||||
<div style="background: #d4edda; border: 1px solid #c3e6cb; border-radius: 5px; padding: 15px; text-align: center;">
|
||||
<p style="margin: 0; font-size: 110%; color: #155724;">
|
||||
<strong>✅ No failed notifications</strong> - All notifications either succeeded or are still being retried.
|
||||
</p>
|
||||
</div>
|
||||
|
||||
<div id="failed-notifications-list">
|
||||
{% for notification in failed_notifications %}
|
||||
<div class="failed-notification-item" style="border: 1px solid #ddd; padding: 15px; margin-bottom: 15px; border-radius: 5px; background: #fafafa;">
|
||||
{% if notification.timestamp %}
|
||||
<div style="margin-bottom: 10px; font-size: 90%; color: #dc3545;">
|
||||
<strong>⚠️ Failed:</strong> {{ notification.timestamp_formatted }}
|
||||
{% if notification.days_ago is defined %}
|
||||
<span style="color: #666;">({{ notification.days_ago }} day{{ 's' if notification.days_ago != 1 else '' }} ago)</span>
|
||||
{% endif %}
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
<div style="margin-bottom: 10px; font-size: 85%; color: #666;">
|
||||
<strong>Task ID:</strong> <code style="background: #eee; padding: 2px 4px; border-radius: 3px; font-size: 80%;">{{ notification.task_id }}</code>
|
||||
</div>
|
||||
|
||||
{% if notification.notification_data and notification.notification_data.get('watch_url') %}
|
||||
<div style="margin-bottom: 10px;">
|
||||
<strong>Watch URL:</strong>
|
||||
<a href="{{ notification.notification_data.get('watch_url') }}" target="_blank" style="word-break: break-all;">
|
||||
{{ notification.notification_data.get('watch_url') }}
|
||||
</a>
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
{% if notification.notification_data and notification.notification_data.get('uuid') %}
|
||||
<div style="margin-bottom: 10px; font-size: 90%;">
|
||||
<strong>Watch UUID:</strong> <code style="background: #eee; padding: 2px 6px; border-radius: 3px;">{{ notification.notification_data.get('uuid') }}</code>
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
{% if notification.retry_attempts %}
|
||||
<div style="margin-bottom: 15px;">
|
||||
<strong>Retry Attempts:</strong>
|
||||
<div style="margin-top: 5px;">
|
||||
{% for attempt in notification.retry_attempts %}
|
||||
<details style="background: #f8f9fa; border: 1px solid #dee2e6; border-radius: 3px; padding: 10px; margin-bottom: 5px;">
|
||||
<summary style="cursor: pointer; font-weight: bold; font-size: 90%;">
|
||||
Attempt #{{ attempt.attempt_number }} - {{ attempt.timestamp_formatted }}
|
||||
{% if attempt.will_retry %}
|
||||
<span style="color: #0066cc;">→ Will retry</span>
|
||||
{% else %}
|
||||
<span style="color: #dc3545;">→ Final attempt</span>
|
||||
{% endif %}
|
||||
</summary>
|
||||
<div style="margin-top: 10px;">
|
||||
<pre style="background: #fff; padding: 8px; border-radius: 3px; border: 1px solid #dee2e6; margin: 0; white-space: pre-wrap; word-wrap: break-word; font-size: 80%;">{{ attempt.error }}</pre>
|
||||
</div>
|
||||
</details>
|
||||
{% endfor %}
|
||||
</div>
|
||||
</div>
|
||||
{% else %}
|
||||
<div style="margin-bottom: 10px;">
|
||||
<strong>Error:</strong>
|
||||
<pre style="background: #fff3cd; padding: 10px; border-radius: 3px; border: 1px solid #ffc107; margin: 5px 0 0 0; white-space: pre-wrap; word-wrap: break-word; font-size: 85%;">{{ notification.error }}</pre>
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
<form method="POST" action="{{ url_for('settings.retry_notification', task_id=notification.task_id) }}" style="margin-top: 10px;">
|
||||
<button type="submit" class="pure-button pure-button-primary" style="font-size: 90%;">
|
||||
<span class="icon-repeat"></span> Retry This Notification
|
||||
</button>
|
||||
</form>
|
||||
</div>
|
||||
{% endfor %}
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{% endblock %}
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user