From 6985f32fe907fae0301a36f8dcaf4a49457fb241 Mon Sep 17 00:00:00 2001 From: AIIrondev Date: Mon, 17 Aug 2026 15:49:51 +0200 Subject: [PATCH] fix(scheduler): eliminate worker race conditions using MongoDB lock - Replace fragile .scheduler_lock file mechanism with atomic MongoDB operations - Prevent multiple background schedulers from starting across web workers - Add heartbeat task to renew lock periodically and handle crash recovery - Release lock cleanly on application shutdown --- Web/app.py | 131 +++++++++++++++++++++++++++++++++++------------------ 1 file changed, 86 insertions(+), 45 deletions(-) diff --git a/Web/app.py b/Web/app.py index 26ff09f..a55270c 100755 --- a/Web/app.py +++ b/Web/app.py @@ -1453,7 +1453,7 @@ def update_appointment_statuses(): - Geplante Termine, die aktiviert werden sollten - Aktive Termine, die beendet werden sollten """ - current_time = datetime.datetime.now() + current_time = datetime.datetime.now(datetime.timezone.utc) try: # Hole alle Termine mit Status 'planned' oder 'active' @@ -1585,6 +1585,9 @@ def update_appointment_statuses(): app.logger.warning( f"Failed to create activation notification for {appointment.get('_id')}: {notif_err}" ) + elif it.is_library_item(appointment.get('Item')): + # Introduction of an messaging system and a Mahnstufen implementation for the library Book bookings after the designatet time. + pass client.close() @@ -1596,78 +1599,116 @@ def update_appointment_statuses(): except Exception as e: app.logger.error(f"Automatic appointment status update failed: {e}") -# Schedule jobs - only start scheduler if this is the main process or a single-worker deployment -# This prevents race conditions in multi-worker Gunicorn environments + +# Initialize scheduler instances scheduler = BackgroundScheduler() _scheduler_initialized = False +_scheduler_worker_id = None # Tracks the unique ID of the worker holding the lock + def _initialize_scheduler(): - """Initialize the background scheduler in a safe way for multi-worker deployments.""" - global _scheduler_initialized - if _scheduler_initialized or not cfg.SCHEDULER_ENABLED: + """Initialize the background scheduler safely using an atomic MongoDB lock.""" + global _scheduler_initialized, _scheduler_worker_id + if _scheduler_initialized or not getattr(cfg, 'SCHEDULER_ENABLED', False): return - + try: - # For multi-worker Gunicorn, use a lock file to ensure only one instance starts the scheduler - # Clean up any stale lock file from previous runs (older than 5 minutes) - scheduler_lock_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), '.scheduler_lock') - try: - if os.path.exists(scheduler_lock_path): - lock_age = time.time() - os.path.getmtime(scheduler_lock_path) - if lock_age > 300: # 5 minutes - indicates a stale lock from a previous container run - os.remove(scheduler_lock_path) - app.logger.info(f"Removed stale scheduler lock file (age: {lock_age:.0f}s)") - except Exception as e: - app.logger.warning(f"Could not clean up scheduler lock file: {e}") + client = MongoClient(MONGODB_HOST, MONGODB_PORT) + db = client[MONGODB_DB] + locks_col = db['system_locks'] - # Always try to remove lock file on startup (extra safety) - try: - if os.path.exists(scheduler_lock_path): - os.remove(scheduler_lock_path) - app.logger.info("Scheduler lock file removed on startup.") - except Exception as e: - app.logger.warning(f"Could not remove scheduler lock file on startup: {e}") + # 1. Generate a unique ID for this specific worker process + _scheduler_worker_id = str(uuid.uuid4()) + now = datetime.datetime.now(datetime.timezone.utc) + # 2. Ensure the lock document exists (initialize if missing) try: - # Try to create the lock file - only succeeds if it doesn't exist - lock_fd = os.open(scheduler_lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644) - os.close(lock_fd) - should_start = True - except FileExistsError: - should_start = False - app.logger.warning("Scheduler lock exists - another process is already running the scheduler") + locks_col.insert_one({ + '_id': 'scheduler_lock', + 'locked_at': datetime.datetime.min.replace(tzinfo=datetime.timezone.utc), + 'worker_id': None + }) + except DuplicateKeyError: + pass # Document already exists, which is expected - if should_start: + # 3. Try to acquire the lock atomically + # We only acquire if the current lock is older than 5 minutes (stale/crashed worker) + # or if it was explicitly released (1970 init date) + lock_timeout = now - datetime.timedelta(minutes=5) + + acquired = locks_col.find_one_and_update( + { + '_id': 'scheduler_lock', + 'locked_at': {'$lt': lock_timeout} + }, + { + '$set': { + 'locked_at': now, + 'worker_id': _scheduler_worker_id + } + } + ) + + if acquired is not None: + # 4. Lock acquired successfully - this is the master worker scheduler.add_job(func=create_daily_backup, trigger="interval", hours=cfg.BACKUP_INTERVAL_HOURS) scheduler.add_job(func=update_appointment_statuses, trigger="interval", minutes=cfg.SCHEDULER_INTERVAL_MIN) scheduler.add_job(func=create_return_reminders, trigger="interval", minutes=cfg.SCHEDULER_INTERVAL_MIN) scheduler.add_job(func=cleanup_expired_trial_tenants, trigger="interval", hours=1) + + # 5. Add a heartbeat job to keep the lock alive + def renew_lock_heartbeat(): + try: + hb_client = MongoClient(MONGODB_HOST, MONGODB_PORT) + hb_db = hb_client[MONGODB_DB] + hb_db['system_locks'].update_one( + {'_id': 'scheduler_lock', 'worker_id': _scheduler_worker_id}, + {'$set': {'locked_at': datetime.datetime.now(datetime.timezone.utc)}} + ) + hb_client.close() + except Exception as e: + app.logger.error(f"Scheduler heartbeat failed: {e}") + + # Run heartbeat every 2 minutes (safely below the 5-minute timeout) + scheduler.add_job(func=renew_lock_heartbeat, trigger="interval", minutes=2) + scheduler.start() _scheduler_initialized = True - app.logger.info(f"Scheduler started successfully (interval={cfg.SCHEDULER_INTERVAL_MIN} min)") + app.logger.info(f"Scheduler started successfully (Worker ID: {_scheduler_worker_id})") else: - app.logger.info("Scheduler skipped - another worker instance is running it") + app.logger.info("Scheduler skipped - another active worker holds the MongoDB lock") + + client.close() + except Exception as e: - app.logger.error(f"Failed to initialize scheduler: {e}") + app.logger.error(f"Failed to initialize scheduler with MongoDB lock: {e}") _scheduler_initialized = False + # Initialize scheduler on app startup _initialize_scheduler() -# Register shutdown handler to stop scheduler when app is terminated -import atexit + def _shutdown_scheduler(): - if cfg.SCHEDULER_ENABLED and _scheduler_initialized: + """Gracefully shut down the scheduler and release the MongoDB lock.""" + global _scheduler_initialized, _scheduler_worker_id + if getattr(cfg, 'SCHEDULER_ENABLED', False) and _scheduler_initialized: try: - scheduler.shutdown() - lock_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), '.scheduler_lock') - try: - os.remove(lock_path) - except Exception: - pass + scheduler.shutdown(wait=False) + + # Release the lock so a newly spawned worker can immediately take over + client = MongoClient(MONGODB_HOST, MONGODB_PORT) + db = client[MONGODB_DB] + db['system_locks'].update_one( + {'_id': 'scheduler_lock', 'worker_id': _scheduler_worker_id}, + {'$set': {'locked_at': datetime.datetime.min.replace(tzinfo=datetime.timezone.utc)}} + ) + client.close() + app.logger.info(f"Scheduler shut down and lock released (Worker ID: {_scheduler_worker_id})") except Exception as e: app.logger.error(f"Error during scheduler shutdown: {e}") + atexit.register(_shutdown_scheduler) """-------------------------------------------------------------File Upload Validation----------------------------------------------------------------------------- """