diff --git a/changedetectionio/async_update_worker.py b/changedetectionio/async_update_worker.py index afb59901c..cabed7c50 100644 --- a/changedetectionio/async_update_worker.py +++ b/changedetectionio/async_update_worker.py @@ -89,11 +89,16 @@ async def async_update_worker(worker_id, q, notification_q, app, datastore, exec continue uuid = queued_item_data.item.get('uuid') - # RACE CONDITION FIX: Check if this UUID is already being processed by another worker + + # RACE CONDITION FIX: Atomically claim this UUID for processing from changedetectionio import worker_handler from changedetectionio.queuedWatchMetaData import PrioritizedItem - if worker_handler.is_watch_running_by_another_worker(uuid, worker_id): - logger.trace(f"Worker {worker_id} detected UUID {uuid} already being processed by another worker - deferring") + + # Try to claim the UUID atomically - prevents duplicate processing + if not worker_handler.claim_uuid_for_processing(uuid, worker_id): + # Already being processed by another worker + logger.trace(f"Worker {worker_id} detected UUID {uuid} already being processed - deferring") + # Sleep to avoid tight loop and give the other worker time to finish await asyncio.sleep(DEFER_SLEEP_TIME_ALREADY_QUEUED) @@ -105,9 +110,6 @@ async def async_update_worker(worker_id, q, notification_q, app, datastore, exec continue fetch_start_time = round(time.time()) - - # Mark this UUID as being processed by this worker - worker_handler.set_uuid_processing(uuid, worker_id=worker_id, processing=True) try: if uuid in list(datastore.data['watching'].keys()) and datastore.data['watching'][uuid].get('url'): @@ -487,8 +489,8 @@ async def async_update_worker(worker_id, q, notification_q, app, datastore, exec except Exception as e: logger.error(f"Exception while cleaning/quit after calling browser: {e}") try: - # Mark UUID as no longer being processed by this worker - worker_handler.set_uuid_processing(uuid, worker_id=worker_id, processing=False) + # Release UUID from processing (thread-safe) + worker_handler.release_uuid_from_processing(uuid, worker_id=worker_id) # Send completion signal if watch: diff --git a/changedetectionio/store.py b/changedetectionio/store.py index 330c59486..296faf787 100644 --- a/changedetectionio/store.py +++ b/changedetectionio/store.py @@ -338,7 +338,6 @@ class ChangeDetectionStore: self.needs_write_urgent = True def add_watch(self, url, tag='', extras=None, tag_uuids=None, write_to_disk_now=True): - import requests if extras is None: extras = {} @@ -349,6 +348,8 @@ class ChangeDetectionStore: # Was it a share link? try to fetch the data if (url.startswith("https://changedetection.io/share/")): + import requests + try: r = requests.request(method="GET", url=url, diff --git a/changedetectionio/tests/util.py b/changedetectionio/tests/util.py index 2b8dda989..8dae12fb1 100644 --- a/changedetectionio/tests/util.py +++ b/changedetectionio/tests/util.py @@ -166,12 +166,14 @@ def wait_for_all_checks(client=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 - time.sleep(0.3) 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 addcb6a51..208a7eb95 100644 --- a/changedetectionio/worker_handler.py +++ b/changedetectionio/worker_handler.py @@ -17,6 +17,7 @@ worker_threads = [] # List of WorkerThread objects # Track currently processing UUIDs for async workers - maps {uuid: worker_id} currently_processing_uuids = {} +_uuid_processing_lock = threading.Lock() # Protects currently_processing_uuids # Configuration - async workers only USE_ASYNC_WORKERS = True @@ -207,31 +208,80 @@ def get_worker_count(): def get_running_uuids(): """Get list of UUIDs currently being processed by async workers""" - return list(currently_processing_uuids.keys()) + with _uuid_processing_lock: + return list(currently_processing_uuids.keys()) + + +def claim_uuid_for_processing(uuid, worker_id): + """ + Atomically check if UUID is available and claim it for processing. + + This is thread-safe and prevents race conditions where multiple workers + try to process the same UUID simultaneously. + + Args: + uuid: The watch UUID to claim + worker_id: The ID of the worker claiming this UUID + + Returns: + True if successfully claimed (UUID was not being processed) + False if already being processed by another worker + """ + with _uuid_processing_lock: + if uuid in currently_processing_uuids: + # Already being processed by another worker + return False + # Claim it atomically + currently_processing_uuids[uuid] = worker_id + logger.debug(f"Worker {worker_id} claimed UUID: {uuid}") + return True + + +def release_uuid_from_processing(uuid, worker_id): + """ + Release a UUID from processing (thread-safe). + + Args: + uuid: The watch UUID to release + worker_id: The ID of the worker releasing this UUID + """ + with _uuid_processing_lock: + # Only remove if this worker owns it (defensive) + if currently_processing_uuids.get(uuid) == worker_id: + currently_processing_uuids.pop(uuid, None) + logger.debug(f"Worker {worker_id} released UUID: {uuid}") + else: + logger.warning(f"Worker {worker_id} tried to release UUID {uuid} but doesn't own it (owned by {currently_processing_uuids.get(uuid, 'nobody')})") def set_uuid_processing(uuid, worker_id=None, processing=True): - """Mark a UUID as being processed or completed by a specific worker""" - global currently_processing_uuids + """ + Mark a UUID as being processed or completed by a specific worker. + + DEPRECATED: Use claim_uuid_for_processing() and release_uuid_from_processing() instead. + This function is kept for backward compatibility but doesn't provide atomic check-and-set. + """ if processing: - currently_processing_uuids[uuid] = worker_id - logger.debug(f"Worker {worker_id} started processing UUID: {uuid}") + with _uuid_processing_lock: + currently_processing_uuids[uuid] = worker_id + logger.debug(f"Worker {worker_id} started processing UUID: {uuid}") else: - currently_processing_uuids.pop(uuid, None) - logger.debug(f"Worker {worker_id} finished processing UUID: {uuid}") + release_uuid_from_processing(uuid, worker_id) def is_watch_running(watch_uuid): """Check if a specific watch is currently being processed by any worker""" - return watch_uuid in currently_processing_uuids + with _uuid_processing_lock: + return watch_uuid in currently_processing_uuids def is_watch_running_by_another_worker(watch_uuid, current_worker_id): """Check if a specific watch is currently being processed by a different worker""" - if watch_uuid not in currently_processing_uuids: - return False - processing_worker_id = currently_processing_uuids[watch_uuid] - return processing_worker_id != current_worker_id + with _uuid_processing_lock: + if watch_uuid not in currently_processing_uuids: + return False + processing_worker_id = currently_processing_uuids[watch_uuid] + return processing_worker_id != current_worker_id def queue_item_async_safe(update_q, item, silent=False):