Improved auto-scan task initialization and status tracking.

- Added autoscan status tracking in pipeline
- Ensured auto-scan runs only once per startup

lightrag/api/lightrag_server.py

@@ -141,23 +141,25 @@ def create_app(args):
         try:
             # Initialize database connections
             await rag.initialize_storages()
-            await initialize_pipeline_status()
 
+            await initialize_pipeline_status()
+            pipeline_status = await get_namespace_data("pipeline_status")
+
+            should_start_autoscan = False
+            async with get_pipeline_status_lock():
                 # Auto scan documents if enabled
                 if args.auto_scan_at_startup:
-                # Check if a task is already running (with lock protection)
+                    if not pipeline_status.get("autoscanned", False):
-                pipeline_status = await get_namespace_data("pipeline_status")
+                        pipeline_status["autoscanned"] = True
-                should_start_task = False
+                        should_start_autoscan = True
-                async with get_pipeline_status_lock():
+
-                    if not pipeline_status.get("busy", False):
+            # Only run auto scan when no other process started it first
-                        should_start_task = True
+            if should_start_autoscan:
-                # Only start the task if no other task is running
-                if should_start_task:
                 # Create background task
                 task = asyncio.create_task(run_scanning_process(rag, doc_manager))
                 app.state.background_tasks.add(task)
                 task.add_done_callback(app.state.background_tasks.discard)
-                    logger.info("Auto scan task started at startup.")
+                logger.info(f"Process {os.getpid()} auto scan task started at startup.")
 
             ASCIIColors.green("\nServer is ready to accept connections! 🚀\n")
 

lightrag/kg/shared_storage.py

@@ -286,6 +286,7 @@ async def initialize_pipeline_status():
         history_messages = _manager.list() if is_multiprocess else []
         pipeline_namespace.update(
             {
+                "autoscanned": False,  # Auto-scan started
                 "busy": False,  # Control concurrent processes
                 "job_name": "Default Job",  # Current job name (indexing files/indexing texts)
                 "job_start": None,  # Job start time