diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 1eeb511..4ee1da3 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -63,7 +63,7 @@ deploy: [Service] WorkingDirectory=${APP_DIR} - Environment="PATH=${VENV_DIR}/bin" "DISABLE_SCHEDULER=true" + Environment="PATH=${VENV_DIR}/bin" ExecStart=${VENV_DIR}/bin/uwsgi \\ --master \\ --enable-threads \\ diff --git a/app.py b/app.py index ac7ae53..328a760 100644 --- a/app.py +++ b/app.py @@ -641,12 +641,12 @@ def dispatch_namespace_scheduler_jobs(): log.info(f"Scheduler dispatching jobs to {len(namespace_dbs)} namespaces") - # Process each namespace in a separate thread to avoid blocking + # Process each namespace in a separate thread with quick queries import concurrent.futures with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: futures = [] for namespace_id in namespace_dbs: - future = executor.submit(process_namespace_scheduled_actions, namespace_id) + future = executor.submit(process_namespace_scheduled_actions_fast, namespace_id) futures.append(future) # Wait for all namespace jobs to complete (with timeout) @@ -706,6 +706,93 @@ def process_namespace_scheduled_actions(namespace_id): namespace_engine.dispose() +def process_namespace_scheduled_actions_fast(namespace_id): + """Process scheduled actions with lightning-fast micro-transactions.""" + namespace_db_url = get_namespace_db_url(namespace_id) + + try: + # Step 1: Quick query to find due actions (read-only, minimal lock) + due_action_ids = [] + engine = create_engine(namespace_db_url, connect_args={"check_same_thread": False}, poolclass=StaticPool) + + # Check if table exists first + from sqlalchemy import inspect + inspector = inspect(engine) + if "scheduled_actions" not in inspector.get_table_names(): + engine.dispose() + return + + # Quick read-only query for due actions + SessionFactory = sessionmaker(bind=engine) + with SessionFactory() as session: + now = datetime.datetime.utcnow() + due_actions = session.query(ScheduledAction.id).filter( + ScheduledAction.status == "pending", + ScheduledAction.scheduled_date <= now, + ).all() + due_action_ids = [action.id for action in due_actions] + engine.dispose() + + if not due_action_ids: + return + + log.info(f"Processing {len(due_action_ids)} due actions in namespace {namespace_id}") + + # Step 2: Process each action in its own micro-transaction + for action_id in due_action_ids: + try: + process_single_action_fast(namespace_id, action_id) + except Exception as e: + log.error(f"Failed to process action {action_id}: {e}") + + except Exception as e: + log.error(f"Error processing namespace {namespace_id}: {e}") + + +def process_single_action_fast(namespace_id, action_id): + """Process a single action with minimal transaction time.""" + namespace_db_url = get_namespace_db_url(namespace_id) + engine = create_engine(namespace_db_url, connect_args={"check_same_thread": False}, poolclass=StaticPool) + + try: + SessionFactory = sessionmaker(bind=engine) + with SessionFactory() as session: + # Get action and media in one query + action = session.query(ScheduledAction).filter_by(id=action_id).first() + if not action or action.status != "pending": + return + + media = session.query(Media).filter_by(id=action.media_id).first() + if not media: + action.status = "failed" + action.completed_at = datetime.datetime.utcnow() + session.commit() + return + + # Perform the action + if action.action_type == "delete": + session.delete(media) + log.info(f"Scheduled delete executed for media {media.filename}") + elif action.action_type == "set_public": + media.visibility = "public" + log.info(f"Scheduled set_public executed for media {media.filename}") + elif action.action_type == "set_private": + media.visibility = "private" + log.info(f"Scheduled set_private executed for media {media.filename}") + elif action.action_type == "set_unlisted": + media.visibility = "unlisted" + log.info(f"Scheduled set_unlisted executed for media {media.filename}") + + action.status = "completed" + action.completed_at = datetime.datetime.utcnow() + session.commit() + + except Exception as e: + log.error(f"Error processing single action {action_id}: {e}") + finally: + engine.dispose() + + def process_single_scheduled_action(dbsession, action): """Process a single scheduled action with minimal transaction time.""" try: @@ -777,21 +864,29 @@ def scheduler_worker(): time.sleep(1) -def start_scheduler(): - """Start the background scheduler.""" - global scheduler_thread, scheduler_running - +def check_and_run_scheduler(): + """Check for due scheduled actions and run them immediately (on-demand).""" # Check if scheduler is disabled via environment variable disable_scheduler = os.environ.get('DISABLE_SCHEDULER', '').lower() in ('true', '1', 'yes') if disable_scheduler: - log.info("Scheduler disabled by DISABLE_SCHEDULER environment variable") return - if not scheduler_running: - scheduler_running = True - scheduler_thread = threading.Thread(target=scheduler_worker, daemon=True) - scheduler_thread.start() - log.info("Scheduler started") + try: + # Simple time-based throttling - only run once per minute max + import time + current_time = time.time() + if not hasattr(check_and_run_scheduler, 'last_run'): + check_and_run_scheduler.last_run = 0 + + if current_time - check_and_run_scheduler.last_run < 60: + return # Skip if ran within last minute + + check_and_run_scheduler.last_run = current_time + log.debug("Running on-demand scheduler check") + dispatch_namespace_scheduler_jobs() + + except Exception as e: + log.error(f"On-demand scheduler error: {e}") def stop_scheduler():