v1.2.20: remaining Codex fixes — _op_results lock, advisory lock for scenarios, _ensure_schema logging, stale async generation token, validate-cfs explicit JSONDecodeError

site/db/scenario_defs.py

@@ -209,27 +209,56 @@ def delete_definition(def_id, client_id, stand):
 def lock_check(client_id, stand):
     """Проверить что нет активного RUNNING-запуска сценария.
 
-    Используется перед запуском нового сценария — предотвращает
-    одновременный запуск двух сценариев одним пользователем.
+    Использует pg_try_advisory_lock для АТОМАРНОЙ проверки-и-захвата.
+    Это устраняет race condition между lock_check и INSERT scenario_runs:
+    если advisory lock взят — никто другой не сможет запустить сценарий
+    для этого же client_id+stand, пока мы его не отпустим.
 
     Returns:
-        True  — можно запускать (нет RUNNING)
-        False — нельзя (уже есть RUNNING) → 409 Conflict"""
+        True  — можно запускать (lock взят)
+        False — нельзя (другой процесс держит lock) → 409 Conflict"""
     conn = get_conn()
     if not conn:
         return True  # без БД — разрешаем (fallback)
     try:
         cur = conn.cursor()
-        cur.execute("""
-            SELECT id FROM scenario_runs
-            WHERE client_id = %s AND stand = %s AND status = 'RUNNING'
-            LIMIT 1
-        """, (client_id, stand))
-        row = cur.fetchone()
+        # hashtext даёт стабильный int из строки — одинаковый во всех сессиях
+        lock_key = f"scenario:{client_id}:{stand}"
+        cur.execute("SELECT pg_try_advisory_lock(hashtext(%s))", (lock_key,))
+        acquired = cur.fetchone()[0]
         cur.close()
-        return row is None  # None = нет RUNNING = можно запускать
+        if not acquired:
+            put_conn(conn)
+            return False  # lock уже взят другим процессом
+        # НЕ возвращаем conn в пул! Держим до unlock.
+        # conn будет передан в сценарий и освобождён после завершения.
+        return True
     except Exception as e:
         print(f"[DEFS] lock_check error: {e}", flush=True)
-        return True
+        put_conn(conn)
+        return True  # fallback
+
+
+def unlock_scenario(client_id, stand, conn=None):
+    """Освободить advisory lock после завершения сценария.
+
+    Вызывается из run_scenario (scenario.py) после завершения ВСЕХ шагов.
+    conn — то же соединение, на котором был взят lock (если есть)."""
+    if not conn:
+        conn = get_conn()
+        if not conn:
+            return
+    try:
+        lock_key = f"scenario:{client_id}:{stand}"
+        cur = conn.cursor()
+        cur.execute("SELECT pg_advisory_unlock(hashtext(%s))", (lock_key,))
+        cur.close()
+        conn.commit()
+    except Exception as e:
+        print(f"[DEFS] unlock_scenario error: {e}", flush=True)
+        try:
+            conn.rollback()
+        except Exception:
+            pass
     finally:
         put_conn(conn)