diff --git a/changedetectionio/__init__.py b/changedetectionio/__init__.py index a6d9749ad..267a7ef56 100644 --- a/changedetectionio/__init__.py +++ b/changedetectionio/__init__.py @@ -161,6 +161,8 @@ def main(): print('Recheck on startup:') print(' -r all Queue all watches for recheck on startup') print(' -r UUID,... Queue specific watches (comma-separated UUIDs)') + print(' -r all N Queue all watches, wait for completion, repeat N times') + print(' -r UUID,... N Queue specific watches, wait for completion, repeat N times') print('') print('Batch mode:') print(' -b Run in batch mode (process queue then exit)') @@ -190,6 +192,7 @@ def main(): urls_to_add = [] url_options = {} # Key: index (0-based), Value: dict of options recheck_watches = None # None, 'all', or list of UUIDs + recheck_repeat_count = 1 # Number of times to repeat recheck cycle batch_mode = False # Run once then exit when queue is empty # On Windows, create and use a default path. @@ -235,7 +238,16 @@ def main(): else: # Parse comma-separated list of UUIDs recheck_watches = [uuid.strip() for uuid in recheck_arg.split(',') if uuid.strip()] - i += 2 + + # Check for optional repeat count as third argument + if i + 2 < len(sys.argv) and sys.argv[i + 2].isdigit(): + recheck_repeat_count = int(sys.argv[i + 2]) + if recheck_repeat_count < 1: + print(f'Error: Repeat count must be at least 1, got {recheck_repeat_count}') + sys.exit(2) + i += 3 + else: + i += 2 continue # Handle -b (batch mode - run once and exit) @@ -273,6 +285,8 @@ def main(): print('Recheck on startup:') print(' -r all Queue all watches for recheck on startup') print(' -r UUID,... Queue specific watches (comma-separated UUIDs)') + print(' -r all N Queue all watches, wait for completion, repeat N times') + print(' -r UUID,... N Queue specific watches, wait for completion, repeat N times') print('') print('Batch mode:') print(' -b Run in batch mode (process queue then exit)') @@ -356,7 +370,9 @@ def main(): # isnt there some @thingy to attach to each route to tell it, that this route needs a datastore app_config = { 'datastore_path': datastore_path, - 'batch_mode': batch_mode + 'batch_mode': batch_mode, + 'recheck_watches': recheck_watches, + 'recheck_repeat_count': recheck_repeat_count } if not os.path.isdir(app_config['datastore_path']): @@ -462,46 +478,88 @@ def main(): if batch_mode: from changedetectionio.flask_app import update_q + def queue_watches_for_recheck(datastore, iteration): + """Helper function to queue watches for recheck""" + watches_to_queue = [] + if recheck_watches == 'all': + all_watches = list(datastore.data['watching'].keys()) + if batch_mode and added_watch_uuids and iteration == 1: + # Only exclude newly added watches on first iteration + watches_to_queue = [uuid for uuid in all_watches if uuid not in added_watch_uuids] + else: + watches_to_queue = all_watches + logger.info(f"Batch mode (iteration {iteration}): Queuing all {len(watches_to_queue)} watches") + elif recheck_watches: + watches_to_queue = recheck_watches + logger.info(f"Batch mode (iteration {iteration}): Queuing {len(watches_to_queue)} specific watches") + + queued_count = 0 + for watch_uuid in watches_to_queue: + if watch_uuid in datastore.data['watching']: + try: + worker_handler.queue_item_async_safe( + update_q, + queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': watch_uuid}) + ) + queued_count += 1 + except Exception as e: + logger.error(f"Failed to queue watch {watch_uuid}: {e}") + else: + logger.warning(f"Watch UUID not found in datastore: {watch_uuid}") + logger.success(f"Batch mode (iteration {iteration}): Successfully queued {queued_count} watches") + return queued_count + def batch_mode_monitor(): - """Monitor queue and workers, shutdown when all work is complete""" + """Monitor queue and workers, shutdown or repeat when work is complete""" import time - logger.info("Batch mode: Waiting for all queued items to complete...") + # Track iterations if repeat mode is enabled + current_iteration = 1 + total_iterations = recheck_repeat_count if recheck_watches and recheck_repeat_count > 1 else 1 + + if total_iterations > 1: + logger.info(f"Batch mode: Will repeat recheck {total_iterations} times") + else: + logger.info("Batch mode: Waiting for all queued items to complete...") # Wait a bit for workers to start processing time.sleep(3) - idle_start = None - idle_threshold = 3 # Seconds to wait after queue is empty + try: + while current_iteration <= total_iterations: + logger.info(f"Batch mode: Waiting for iteration {current_iteration}/{total_iterations} to complete...") - while True: - try: - queue_size = update_q.qsize() - running_uuids = worker_handler.get_running_uuids() - running_count = len(running_uuids) + # Use the shared wait_for_all_checks function + completed = worker_handler.wait_for_all_checks(update_q, timeout=300) - logger.info(f"Batch mode: Queue size: {queue_size}, Running workers: {running_count}, Running UUIDs: {running_uuids}") + if not completed: + logger.warning(f"Batch mode: Iteration {current_iteration} timed out after 300 seconds") - if queue_size == 0 and running_count == 0: - if idle_start is None: - idle_start = time.time() - logger.info(f"Batch mode: Queue empty and workers idle, waiting {idle_threshold}s before shutdown...") - elif time.time() - idle_start >= idle_threshold: - logger.success("Batch mode: All work completed, initiating shutdown") - # Trigger shutdown - import os, signal - os.kill(os.getpid(), signal.SIGTERM) - return + logger.success(f"Batch mode: Iteration {current_iteration}/{total_iterations} completed") + + # Check if we need to repeat + if current_iteration < total_iterations: + logger.info(f"Batch mode: Starting iteration {current_iteration + 1}...") + current_iteration += 1 + + # Re-queue watches for next iteration + queue_watches_for_recheck(datastore, current_iteration) + + # Brief pause before continuing + time.sleep(2) else: - if idle_start is not None: - logger.info(f"Batch mode: Activity detected (queue: {queue_size}, running: {running_count})") - idle_start = None + # All iterations complete + logger.success(f"Batch mode: All {total_iterations} iterations completed, initiating shutdown") + # Trigger shutdown + import os, signal + os.kill(os.getpid(), signal.SIGTERM) + return - time.sleep(2) - - except Exception as e: - logger.error(f"Batch mode monitor error: {e}") - time.sleep(2) + except Exception as e: + logger.error(f"Batch mode monitor error: {e}") + logger.error(f"Initiating emergency shutdown") + import os, signal + os.kill(os.getpid(), signal.SIGTERM) # Start monitor in background thread monitor_thread = threading.Thread(target=batch_mode_monitor, daemon=True, name="BatchModeMonitor") diff --git a/changedetectionio/tests/util.py b/changedetectionio/tests/util.py index 8dae12fb1..a531d4a54 100644 --- a/changedetectionio/tests/util.py +++ b/changedetectionio/tests/util.py @@ -140,40 +140,13 @@ def delete_all_watches(client=None): def wait_for_all_checks(client=None): """ Waits until the queue is empty and workers are idle. - Much faster than the original with adaptive timing. + Delegates to worker_handler.wait_for_all_checks for shared logic. """ from changedetectionio.flask_app import update_q as global_update_q from changedetectionio import worker_handler - empty_since = None - attempt = 0 - max_attempts = 150 # Still reasonable upper bound - while attempt < max_attempts: - # Start with fast checks, slow down if needed - if attempt < 10: - time.sleep(0.2) # Very fast initial checks - elif attempt < 30: - time.sleep(0.4) # Medium speed - else: - time.sleep(0.8) # Slower for persistent issues - - q_length = global_update_q.qsize() - running_uuids = worker_handler.get_running_uuids() - any_workers_busy = len(running_uuids) > 0 - - if q_length == 0 and not any_workers_busy: - if empty_since is None: - empty_since = time.time() - # Brief stabilization period for async workers - elif time.time() - empty_since >= 0.3: - # Add small buffer for filesystem operations to complete - # This ensures history blobs and HTML snapshots are fully written - time.sleep(0.2) - break - else: - empty_since = None - - attempt += 1 + # Use the shared wait logic from worker_handler + return worker_handler.wait_for_all_checks(global_update_q, timeout=150) def wait_for_watch_history(client, min_history_count=2, timeout=10): """ diff --git a/changedetectionio/worker_handler.py b/changedetectionio/worker_handler.py index 208a7eb95..51aa2eccb 100644 --- a/changedetectionio/worker_handler.py +++ b/changedetectionio/worker_handler.py @@ -431,6 +431,53 @@ def get_worker_status(): } +def wait_for_all_checks(update_q, timeout=150): + """ + Wait for queue to be empty and all workers to be idle. + + Args: + update_q: The update queue to monitor + timeout: Maximum wait time in seconds (default 150 = 150 iterations * 0.2-0.8s) + + Returns: + bool: True if all checks completed, False if timeout + """ + import time + empty_since = None + attempt = 0 + max_attempts = timeout + + while attempt < max_attempts: + # Adaptive sleep - start fast, slow down if needed + if attempt < 10: + sleep_time = 0.2 # Very fast initial checks + elif attempt < 30: + sleep_time = 0.4 # Medium speed + else: + sleep_time = 0.8 # Slower for persistent issues + + time.sleep(sleep_time) + + q_length = update_q.qsize() + running_uuids = get_running_uuids() + any_workers_busy = len(running_uuids) > 0 + + if q_length == 0 and not any_workers_busy: + if empty_since is None: + empty_since = time.time() + # Brief stabilization period for async workers + elif time.time() - empty_since >= 0.3: + # Add small buffer for filesystem operations to complete + time.sleep(0.2) + return True + else: + empty_since = None + + attempt += 1 + + return False # Timeout + + def check_worker_health(expected_count, update_q=None, notification_q=None, app=None, datastore=None): """ Check if the expected number of async workers are running and restart any missing ones.