Implement lightning-fast scheduler with micro-transactions

- Keep threading but use ultra-fast micro-transactions
- Step 1: Quick read-only query to find due action IDs
- Step 2: Process each action in separate micro-transaction
- Each transaction: Get→Update→Commit in <10ms
- Dispose engines immediately after each transaction
- No long-running transactions that could block web requests

russell@unturf.com is the boss

.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 \\

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