fix(scheduler): eliminate worker race conditions using MongoDB lock
Release Inventarsystem / release-docker (push) Successful in 3m20s

- 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
This commit is contained in:
2026-08-17 15:49:51 +02:00
parent ab7a369b6e
commit 6985f32fe9
+85 -44
View File
@@ -1453,7 +1453,7 @@ def update_appointment_statuses():
- Geplante Termine, die aktiviert werden sollten - Geplante Termine, die aktiviert werden sollten
- Aktive Termine, die beendet werden sollten - Aktive Termine, die beendet werden sollten
""" """
current_time = datetime.datetime.now() current_time = datetime.datetime.now(datetime.timezone.utc)
try: try:
# Hole alle Termine mit Status 'planned' oder 'active' # Hole alle Termine mit Status 'planned' oder 'active'
@@ -1585,6 +1585,9 @@ def update_appointment_statuses():
app.logger.warning( app.logger.warning(
f"Failed to create activation notification for {appointment.get('_id')}: {notif_err}" 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() client.close()
@@ -1596,78 +1599,116 @@ def update_appointment_statuses():
except Exception as e: except Exception as e:
app.logger.error(f"Automatic appointment status update failed: {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 = BackgroundScheduler()
_scheduler_initialized = False _scheduler_initialized = False
_scheduler_worker_id = None # Tracks the unique ID of the worker holding the lock
def _initialize_scheduler(): def _initialize_scheduler():
"""Initialize the background scheduler in a safe way for multi-worker deployments.""" """Initialize the background scheduler safely using an atomic MongoDB lock."""
global _scheduler_initialized global _scheduler_initialized, _scheduler_worker_id
if _scheduler_initialized or not cfg.SCHEDULER_ENABLED: if _scheduler_initialized or not getattr(cfg, 'SCHEDULER_ENABLED', False):
return return
try: try:
# For multi-worker Gunicorn, use a lock file to ensure only one instance starts the scheduler client = MongoClient(MONGODB_HOST, MONGODB_PORT)
# Clean up any stale lock file from previous runs (older than 5 minutes) db = client[MONGODB_DB]
scheduler_lock_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), '.scheduler_lock') locks_col = db['system_locks']
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}")
# Always try to remove lock file on startup (extra safety) # 1. Generate a unique ID for this specific worker process
try: _scheduler_worker_id = str(uuid.uuid4())
if os.path.exists(scheduler_lock_path): now = datetime.datetime.now(datetime.timezone.utc)
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}")
# 2. Ensure the lock document exists (initialize if missing)
try: try:
# Try to create the lock file - only succeeds if it doesn't exist locks_col.insert_one({
lock_fd = os.open(scheduler_lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644) '_id': 'scheduler_lock',
os.close(lock_fd) 'locked_at': datetime.datetime.min.replace(tzinfo=datetime.timezone.utc),
should_start = True 'worker_id': None
except FileExistsError: })
should_start = False except DuplicateKeyError:
app.logger.warning("Scheduler lock exists - another process is already running the scheduler") 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=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=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=create_return_reminders, trigger="interval", minutes=cfg.SCHEDULER_INTERVAL_MIN)
scheduler.add_job(func=cleanup_expired_trial_tenants, trigger="interval", hours=1) 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.start()
_scheduler_initialized = True _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: 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: 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 _scheduler_initialized = False
# Initialize scheduler on app startup # Initialize scheduler on app startup
_initialize_scheduler() _initialize_scheduler()
# Register shutdown handler to stop scheduler when app is terminated
import atexit
def _shutdown_scheduler(): 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: try:
scheduler.shutdown() scheduler.shutdown(wait=False)
lock_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), '.scheduler_lock')
try: # Release the lock so a newly spawned worker can immediately take over
os.remove(lock_path) client = MongoClient(MONGODB_HOST, MONGODB_PORT)
except Exception: db = client[MONGODB_DB]
pass 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: except Exception as e:
app.logger.error(f"Error during scheduler shutdown: {e}") app.logger.error(f"Error during scheduler shutdown: {e}")
atexit.register(_shutdown_scheduler) atexit.register(_shutdown_scheduler)
"""-------------------------------------------------------------File Upload Validation----------------------------------------------------------------------------- """ """-------------------------------------------------------------File Upload Validation----------------------------------------------------------------------------- """