Better cross-worker watch UUID management

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:

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,

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):
     """

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):