попытка фикса worker overflow

.env.example

@@ -1,5 +1,10 @@
 # ── Database ──────────────────────────────────────────────────────────────────
 DATABASE_URL=postgresql+asyncpg://fleet:fleet@localhost:5432/fleet
+DATABASE_POOL_SIZE=10
+DATABASE_MAX_OVERFLOW=20
+DATABASE_POOL_TIMEOUT=30
+TASK_DEVICE_CONCURRENCY=10
+MONITOR_PING_CONCURRENCY=20
 
 # ── Security ──────────────────────────────────────────────────────────────────
 # Generate with: python -c "import secrets; print(secrets.token_hex(32))"

backend/app/config.py

@@ -17,5 +17,12 @@ class Settings(BaseSettings):
 
     SQLALCHEMY_LOG: bool = False
 
+    DATABASE_POOL_SIZE: int = 10
+    DATABASE_MAX_OVERFLOW: int = 20
+    DATABASE_POOL_TIMEOUT: int = 30
+
+    TASK_DEVICE_CONCURRENCY: int = 10
+    MONITOR_PING_CONCURRENCY: int = 20
+
 
 settings = Settings()

backend/app/services/database.py

@@ -8,7 +8,10 @@ engine = create_async_engine(
     settings.DATABASE_URL,
     #echo=settings.DEBUG,
     echo=False,
-    pool_pre_ping=True
+    pool_pre_ping=True,
+    pool_size=settings.DATABASE_POOL_SIZE,
+    max_overflow=settings.DATABASE_MAX_OVERFLOW,
+    pool_timeout=settings.DATABASE_POOL_TIMEOUT,
 )
 
 

backend/app/services/ssh.py

@@ -29,6 +29,7 @@ from app.models.device import Device
 from app.models.object import Object
 from app.models.ssh_key import SSHKey
 from app.services.crypto import decrypt
+from app.services.database import AsyncSessionLocal
 
 logger = logging.getLogger(__name__)
 
@@ -355,7 +356,11 @@ async def run_command(
     async with _global_sem:
         async with _get_object_sem(device.object_id):
             async with _get_device_sem(device.id):
+                if db:
                     options = await build_auth_options(device, obj, db)
+                else:
+                    async with AsyncSessionLocal() as auth_db:
+                        options = await build_auth_options(device, obj, auth_db)
 
                 if not options:
                     # Fallback: no creds configured at all
@@ -389,16 +394,29 @@ async def run_command(
                         )
                         if db:
                             await _save_last_auth(device, db, opt)
+                        else:
+                            async with AsyncSessionLocal() as save_db:
+                                async with save_db.begin():
+                                    current_device = await save_db.get(Device, device.id)
+                                    if current_device:
+                                        await _save_last_auth(current_device, save_db, opt)
                         return result
 
                     # Auth failed (wrong password / key / connection error)
                     # If last_auth method failed, clear it
-                    if i == 0 and opt is last_auth_option and db:
+                    if i == 0 and opt is last_auth_option:
                         device.ssh_last_auth = None
+                        if db:
                             try:
                                 await db.flush()
                             except Exception:
                                 pass
+                        else:
+                            async with AsyncSessionLocal() as clear_db:
+                                async with clear_db.begin():
+                                    current_device = await clear_db.get(Device, device.id)
+                                    if current_device:
+                                        current_device.ssh_last_auth = None
 
                     last_result = result
                     logger.info(

backend/app/services/task.py

@@ -20,11 +20,12 @@ from app.models.rack import Rack
 from app.models.task import Task, TaskEvent, TaskResult
 from app.models.user import User
 from app.plugins.registry import ActionRegistry, action_registry
+from app.config import settings
 from app.services.database import AsyncSessionLocal
 from app.services.sse import SSEEvent, sse_manager
 
 # Max parallel devices within a single task
-_TASK_DEVICE_CONCURRENCY = 20
+_TASK_DEVICE_CONCURRENCY = settings.TASK_DEVICE_CONCURRENCY
 
 
 async def _emit(task_id: str, message: str, level: str = "info") -> None:
@@ -115,13 +116,14 @@ async def _run_on_device(
     params: dict,
     sem: asyncio.Semaphore,
 ) -> bool | None:
-    # Bail out before acquiring the semaphore if task was cancelled
+    async with sem:
+        # Bail out after acquiring the semaphore so large target lists don't
+        # stampede the DB pool with cancellation checks.
         async with AsyncSessionLocal() as db:
             task_row = await db.get(Task, task_id)
             if task_row and task_row.cancel_requested:
                 return None  # None = skipped due to cancellation
 
-    async with sem:
         emit = lambda msg, level="info": _emit(task_id, msg, level)
 
         # Create task_result row
@@ -140,9 +142,7 @@ async def _run_on_device(
                 result_id = tr.id
 
         try:
-            async with AsyncSessionLocal() as action_db:
-                async with action_db.begin():
-                    action_result = await action.execute(device, obj, params, emit, db=action_db)
+            action_result = await action.execute(device, obj, params, emit, db=None)
         except Exception as exc:
             action_result_success = False
             action_stdout = ""

backend/app/workers/monitor.py

@@ -21,13 +21,14 @@ from sqlalchemy import select, update
 
 from app.models.alert import Alert
 from app.models.device import Device
+from app.config import settings
 from app.services.database import AsyncSessionLocal
 from app.services.sse import SSEEvent, sse_manager
 
 # ── Configuration ──────────────────────────────────────────────────────────────
 MONITOR_INTERVAL = 60     # seconds between full check cycles
 PING_TIMEOUT = 3          # seconds per ping attempt
-PING_CONCURRENCY = 30     # max simultaneous pings
+PING_CONCURRENCY = settings.MONITOR_PING_CONCURRENCY  # max simultaneous device checks
 
 MONITOR_CHANNEL = "__monitor__"
 
@@ -98,6 +99,14 @@ async def _check_device(
     sem: asyncio.Semaphore,
 ) -> None:
     async with sem:
+        await _check_device_locked(device_id, ip, object_id)
+
+
+async def _check_device_locked(
+    device_id: int,
+    ip: str,
+    object_id: int,
+) -> None:
     success, ping_ms = await _ping(ip)
 
     now = datetime.now(timezone.utc)