diff --git a/changedetectionio/async_update_worker.py b/changedetectionio/async_update_worker.py index 54f7e603e..03807c26a 100644 --- a/changedetectionio/async_update_worker.py +++ b/changedetectionio/async_update_worker.py @@ -30,7 +30,6 @@ async def async_update_worker(worker_id, q, notification_q, app, datastore): while not app.config.exit.is_set(): update_handler = None watch = None - current_uuid = None try: # Use asyncio wait_for to make queue.get() cancellable @@ -43,16 +42,14 @@ async def async_update_worker(worker_id, q, notification_q, app, datastore): await asyncio.sleep(0.1) continue - uuid = None + uuid = queued_item_data.item.get('uuid') + fetch_start_time = round(time.time()) + + # Mark this UUID as being processed + from changedetectionio import worker_handler + worker_handler.set_uuid_processing(uuid, processing=True) + try: - uuid = queued_item_data.item.get('uuid') - fetch_start_time = round(time.time()) - current_uuid = uuid - - # Mark this UUID as being processed - from changedetectionio import worker_handler - worker_handler.set_uuid_processing(uuid, processing=True) - if uuid in list(datastore.data['watching'].keys()) and datastore.data['watching'][uuid].get('url'): changed_detected = False contents = b'' @@ -354,39 +351,37 @@ async def async_update_worker(worker_id, q, notification_q, app, datastore): datastore.update_watch(uuid=uuid, update_obj={'fetch_time': round(time.time() - fetch_start_time, 3), 'check_count': count}) - current_uuid = None - - # Mark UUID as no longer being processed - worker_handler.set_uuid_processing(uuid, processing=False) - - # Send completion signal - if watch: - logger.info(f"Worker {worker_id} sending completion signal for UUID {watch['uuid']}") - watch_check_update.send(watch_uuid=watch['uuid']) - - update_handler = None - logger.debug(f"Worker {worker_id} completed watch {uuid} in {time.time()-fetch_start_time:.2f}s") - - # Yield control to other coroutines - await asyncio.sleep(0.01) - except Exception as e: logger.error(f"Worker {worker_id} unexpected error processing {uuid}: {e}") logger.error(f"Worker {worker_id} traceback:", exc_info=True) - # Make sure to mark UUID as completed even on error + # Also update the watch with error information + if datastore and uuid in datastore.data['watching']: + datastore.update_watch(uuid=uuid, update_obj={'last_error': f"Worker error: {str(e)}"}) + + finally: + # Always cleanup - this runs whether there was an exception or not if uuid: try: + # Mark UUID as no longer being processed worker_handler.set_uuid_processing(uuid, processing=False) - # Also update the watch with error information - if datastore and uuid in datastore.data['watching']: - datastore.update_watch(uuid=uuid, update_obj={'last_error': f"Worker error: {str(e)}"}) + + # Send completion signal + if watch: + #logger.info(f"Worker {worker_id} sending completion signal for UUID {watch['uuid']}") + watch_check_update.send(watch_uuid=watch['uuid']) + + update_handler = None + logger.debug(f"Worker {worker_id} completed watch {uuid} in {time.time()-fetch_start_time:.2f}s") except Exception as cleanup_error: logger.error(f"Worker {worker_id} error during cleanup: {cleanup_error}") - - current_uuid = None - # Brief pause before continuing to avoid tight error loops - await asyncio.sleep(1.0) + + # Brief pause before continuing to avoid tight error loops (only on error) + if 'e' in locals(): + await asyncio.sleep(1.0) + else: + # Small yield for normal completion + await asyncio.sleep(0.01) # Check if we should exit if app.config.exit.is_set(): diff --git a/changedetectionio/realtime/socket_server.py b/changedetectionio/realtime/socket_server.py index e595ea9bd..9bfa7bf7d 100644 --- a/changedetectionio/realtime/socket_server.py +++ b/changedetectionio/realtime/socket_server.py @@ -8,8 +8,10 @@ from blinker import signal from changedetectionio import strtobool + class SignalHandler: """A standalone class to receive signals""" + def __init__(self, socketio_instance, datastore): self.socketio_instance = socketio_instance self.datastore = datastore @@ -17,18 +19,17 @@ class SignalHandler: # Connect to the watch_check_update signal from changedetectionio.flask_app import watch_check_update as wcc wcc.connect(self.handle_signal, weak=False) -# logger.info("SignalHandler: Connected to signal from direct import") - + # logger.info("SignalHandler: Connected to signal from direct import") + # Connect to the queue_length signal queue_length_signal = signal('queue_length') queue_length_signal.connect(self.handle_queue_length, weak=False) - # logger.info("SignalHandler: Connected to queue_length signal") - + # logger.info("SignalHandler: Connected to queue_length signal") # Create and start the queue update thread using eventlet import eventlet self.polling_emitter_thread = eventlet.spawn(self.polling_emit_running_or_queued_watches) - + # Store the thread reference in socketio for clean shutdown self.socketio_instance.polling_emitter_thread = self.polling_emitter_thread @@ -43,7 +44,7 @@ class SignalHandler: watch = self.datastore.data['watching'].get(watch_uuid) if watch: if app_context: - #note + # note with app_context.app_context(): with app_context.test_request_context(): # Forward to handle_watch_update with the watch parameter @@ -60,17 +61,16 @@ class SignalHandler: try: queue_length = kwargs.get('length', 0) logger.debug(f"SignalHandler: Queue length update received: {queue_length}") - + # Emit the queue size to all connected clients self.socketio_instance.emit("queue_size", { "q_length": queue_length, "event_timestamp": time.time() }) - + except Exception as e: logger.error(f"Socket.IO error in handle_queue_length: {str(e)}") - def polling_emit_running_or_queued_watches(self): """Greenlet that periodically updates the browser/frontend with current state of who is being checked or queued This is because sometimes the browser page could reload (like on clicking on a link) but the data is old @@ -81,7 +81,7 @@ class SignalHandler: from changedetectionio.flask_app import app from changedetectionio import worker_handler watch_check_update = signal('watch_check_update') - + # Use eventlet sleep for non-blocking operation from eventlet import sleep as eventlet_sleep @@ -91,28 +91,22 @@ class SignalHandler: # Run until explicitly stopped while stop_event is None or not stop_event.ready(): try: - # For each running UUID, send a signal, so we update the UI - running_uuids = worker_handler.get_running_uuids() - for uuid in running_uuids: - logger.trace(f"Sending update for {uuid}") - # Send with app_context to ensure proper URL generation - with app.app_context(): - watch_check_update.send(app_context=app, watch_uuid=uuid) - # Yield control back to eventlet after each send to prevent blocking - eventlet_sleep(0.1) # Small sleep to yield control - - # Check if we need to stop in the middle of processing - if stop_event is not None and stop_event.ready(): - break + # Get current running UUIDs from async workers + running_uuids = set(worker_handler.get_running_uuids()) - # Sleep between polling/update cycles - eventlet_sleep(2) + # Send updates for newly running UUIDs + with app.app_context(): + for uuid in running_uuids: + watch_check_update.send(app_context=app, watch_uuid=uuid) + eventlet_sleep(0.01) # Small yield except Exception as e: logger.error(f"Error in queue update greenlet: {str(e)}") # Sleep a bit to avoid flooding logs in case of persistent error eventlet_sleep(0.5) + eventlet_sleep(10) + logger.info("Queue update eventlet greenlet stopped") @@ -141,26 +135,29 @@ def handle_watch_update(socketio, **kwargs): error_texts = watch.compile_error_texts() # Create a simplified watch data object to send to clients + watch_uuid = watch.get('uuid') + watch_data = { - 'checking_now': True if watch.get('uuid') in running_uuids else False, + 'checking_now': True if watch_uuid in running_uuids else False, 'fetch_time': watch.get('fetch_time'), 'has_error': True if error_texts else False, 'last_changed': watch.get('last_changed'), 'last_checked': watch.get('last_checked'), 'error_text': error_texts, 'last_checked_text': _jinja2_filter_datetime(watch), - 'last_changed_text': timeago.format(int(watch['last_changed']), time.time()) if watch.history_n >= 2 and int(watch.get('last_changed', 0)) > 0 else 'Not yet', - 'queued': True if watch.get('uuid') in queue_list else False, + 'last_changed_text': timeago.format(int(watch['last_changed']), time.time()) if watch.history_n >= 2 and int( + watch.get('last_changed', 0)) > 0 else 'Not yet', + 'queued': True if watch_uuid in queue_list else False, 'paused': True if watch.get('paused') else False, 'notification_muted': True if watch.get('notification_muted') else False, 'unviewed': watch.has_unviewed, - 'uuid': watch.get('uuid'), + 'uuid': watch_uuid, 'event_timestamp': time.time() } - errored_count =0 - for uuid, watch in datastore.data['watching'].items(): - if watch.get('last_error'): + errored_count = 0 + for watch_uuid_iter, watch_iter in datastore.data['watching'].items(): + if watch_iter.get('last_error'): errored_count += 1 general_stats = { @@ -169,13 +166,13 @@ def handle_watch_update(socketio, **kwargs): } # Debug what's being emitted - #logger.debug(f"Emitting 'watch_update' event for {watch.get('uuid')}, data: {watch_data}") - + # logger.debug(f"Emitting 'watch_update' event for {watch.get('uuid')}, data: {watch_data}") + # Emit to all clients (no 'broadcast' parameter needed - it's the default behavior) socketio.emit("watch_update", {'watch': watch_data, 'general_stats': general_stats}) - - # Log after successful emit - logger.info(f"Socket.IO: Emitted update for watch {watch.get('uuid')}, Checking now: {watch_data['checking_now']}") + + # Log after successful emit - use watch_data['uuid'] to avoid variable shadowing issues + logger.trace(f"Socket.IO: Emitted update for watch {watch_data['uuid']}, Checking now: {watch_data['checking_now']}") except Exception as e: logger.error(f"Socket.IO error in handle_watch_update: {str(e)}") @@ -190,30 +187,31 @@ def init_socketio(app, datastore): # Restrict SocketIO CORS to same origin by default, can be overridden with env var cors_origins = os.environ.get('SOCKETIO_CORS_ORIGINS', None) - + socketio = SocketIO(app, - async_mode=async_mode, - cors_allowed_origins=cors_origins, # None means same-origin only - logger=strtobool(os.getenv('SOCKETIO_LOGGING', 'False')), - engineio_logger=strtobool(os.getenv('SOCKETIO_LOGGING', 'False'))) + async_mode=async_mode, + cors_allowed_origins=cors_origins, # None means same-origin only + logger=strtobool(os.getenv('SOCKETIO_LOGGING', 'False')), + engineio_logger=strtobool(os.getenv('SOCKETIO_LOGGING', 'False'))) # Set up event handlers logger.info("Socket.IO: Registering connect event handler") + @socketio.on('connect') def handle_connect(): """Handle client connection""" -# logger.info("Socket.IO: CONNECT HANDLER CALLED - Starting connection process") + # logger.info("Socket.IO: CONNECT HANDLER CALLED - Starting connection process") from flask import request from flask_login import current_user from changedetectionio.flask_app import update_q # Access datastore from socketio datastore = socketio.datastore -# logger.info(f"Socket.IO: Current user authenticated: {current_user.is_authenticated if hasattr(current_user, 'is_authenticated') else 'No current_user'}") + # logger.info(f"Socket.IO: Current user authenticated: {current_user.is_authenticated if hasattr(current_user, 'is_authenticated') else 'No current_user'}") # Check if authentication is required and user is not authenticated has_password_enabled = datastore.data['settings']['application'].get('password') or os.getenv("SALTED_PASS", False) -# logger.info(f"Socket.IO: Password enabled: {has_password_enabled}") + # logger.info(f"Socket.IO: Password enabled: {has_password_enabled}") if has_password_enabled and not current_user.is_authenticated: logger.warning("Socket.IO: Rejecting unauthenticated connection") return False # Reject the connection @@ -231,7 +229,7 @@ def init_socketio(app, datastore): logger.info("Socket.IO: Client connected") -# logger.info("Socket.IO: Registering disconnect event handler") + # logger.info("Socket.IO: Registering disconnect event handler") @socketio.on('disconnect') def handle_disconnect(): """Handle client disconnection""" @@ -246,24 +244,23 @@ def init_socketio(app, datastore): # Store the datastore reference on the socketio object for later use socketio.datastore = datastore - + # Create a stop event for our queue update thread using eventlet Event import eventlet.event stop_event = eventlet.event.Event() socketio.stop_event = stop_event - # Add a shutdown method to the socketio object def shutdown(): """Shutdown the SocketIO server gracefully""" try: logger.info("Socket.IO: Shutting down server...") - + # Signal the queue update thread to stop if hasattr(socketio, 'stop_event'): socketio.stop_event.send() logger.info("Socket.IO: Signaled queue update thread to stop") - + # Wait for the greenlet to exit (with timeout) if hasattr(socketio, 'polling_emitter_thread'): try: @@ -275,14 +272,14 @@ def init_socketio(app, datastore): logger.info("Socket.IO: Queue update eventlet greenlet already dead") except Exception as e: logger.error(f"Error killing eventlet greenlet: {str(e)}") - + # Close any remaining client connections - #if hasattr(socketio, 'server'): + # if hasattr(socketio, 'server'): # socketio.server.disconnect() logger.info("Socket.IO: Server shutdown complete") except Exception as e: logger.error(f"Socket.IO error during shutdown: {str(e)}") - + # Attach the shutdown method to the socketio object socketio.shutdown = shutdown