From 5065019ffdb53130c4a6069e23f0cafd00d66dfb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CNaeel=E2=80=9D?= Date: Fri, 31 Jul 2026 18:07:24 +0400 Subject: [PATCH] =?UTF-8?q?v1.2.20:=20remaining=20Codex=20fixes=20?= =?UTF-8?q?=E2=80=94=20=5Fop=5Fresults=20lock,=20advisory=20lock=20for=20s?= =?UTF-8?q?cenarios,=20=5Fensure=5Fschema=20logging,=20stale=20async=20gen?= =?UTF-8?q?eration=20token,=20validate-cfs=20explicit=20JSONDecodeError?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- site/app.py | 2 +- site/db/pool.py | 7 +- site/db/scenario_defs.py | 53 +++++-- site/operations/scenario.py | 253 ++++++++++++++++---------------- site/operations/terraform.py | 14 +- site/routes/api_test.py | 55 ++++--- site/static/js/scenario-form.js | 10 +- 7 files changed, 222 insertions(+), 172 deletions(-) diff --git a/site/app.py b/site/app.py index c479fc8..6177297 100644 --- a/site/app.py +++ b/site/app.py @@ -32,7 +32,7 @@ from routes.api_scenario_defs import bp_defs as api_scenario_defs_bp # Версия — показывается в топбаре UI. Меняется при КАЖДОМ изменении кода. # Нужна для фильтрации истории (пользователь видит только записи своей версии). -VERSION = "1.2.19" +VERSION = "1.2.20" # Flask-приложение с Jinja2-шаблонами из папки templates/ app = Flask(__name__, template_folder="templates", static_folder="static") diff --git a/site/db/pool.py b/site/db/pool.py index b7f4c00..b9d945e 100644 --- a/site/db/pool.py +++ b/site/db/pool.py @@ -58,8 +58,11 @@ def _ensure_schema(): from db.init_db import init_db init_db() _initialized = True - except Exception: - pass # без БД приложение работает (без истории) + except Exception as e: + print(f"[DB] Schema init FAILED: {e}", flush=True) + import traceback + traceback.print_exc() + # Приложение продолжает работу БЕЗ БД (без истории/сценариев) def get_pool(): diff --git a/site/db/scenario_defs.py b/site/db/scenario_defs.py index ec748ab..f7a9628 100644 --- a/site/db/scenario_defs.py +++ b/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) diff --git a/site/operations/scenario.py b/site/operations/scenario.py index 9900abc..fc599f0 100644 --- a/site/operations/scenario.py +++ b/site/operations/scenario.py @@ -32,6 +32,7 @@ from operations.executor import execute_operation from operations.poll import poll_until_done from db.pool import get_conn, put_conn from db.save_run import save_run +from db.scenario_defs import unlock_scenario # Все инстансы созданные сценарием имеют такой префикс в displayName AUTOTEST_PREFIX = "autotest-scenario-" @@ -174,135 +175,139 @@ def run_scenario(client, steps, client_id, stand, user_email, app_version, scena bindings = {} # output_name → instance_uid (новый формат, именованные ссылки) instance_map = {} # service_id → instance_uid (fallback, старый формат) - for i, step in enumerate(steps): - step_num = i + 1 # шаги нумеруются с 1 (для пользователя) - svc_id = step.get("service_id") - op_name = (step.get("operation") or "").strip() - symbolic_params = step.get("params", {}) + try: + for i, step in enumerate(steps): + step_num = i + 1 # шаги нумеруются с 1 (для пользователя) + svc_id = step.get("service_id") + op_name = (step.get("operation") or "").strip() + symbolic_params = step.get("params", {}) - try: - # Валидация базовых полей шага - if not isinstance(svc_id, int) or svc_id <= 0: - raise ValueError(f"Invalid service_id: {svc_id!r}") - if not op_name: - raise ValueError("Empty operation name") - - # ═══ Шаг 1: получить svcOperationId для операции ═══ - detail = get_service_detail(client, svc_id) - ops = detail.get("operations", []) - op = next((o for o in ops if o.get("operation") == op_name), None) - if not op: - raise ValueError(f"Operation '{op_name}' not found for service {svc_id}") - svc_op_id = op["svcOperationId"] - - # ═══ Шаг 2: резолвить параметры (символ. имена → числовые ID) ═══ - tmpl_data = client.get(f"/instanceOperations/default/{svc_op_id}") - cfs_params = tmpl_data.get("svcOperation", {}).get("cfsParams", []) - resolved_params = _resolve_params(cfs_params, symbolic_params) - - # ═══ Шаг 3: резолвинг instance_uid ═══ - is_create = (op_name == "create") - if is_create: - # Для create: instance_uid=None (создаём новый) - # displayName генерируем уникальный: autotest-scenario-{name}-{random6} - instance_uid = None - display_name = f"{AUTOTEST_PREFIX}{scenario_name}-{uuid.uuid4().hex[:6]}" - descr = f"scenario {scenario_name} step {step_num}" - else: - # Для не-create: резолвим instance_uid - instance_uid = _resolve_instance_uid(step, bindings, instance_map) - if not instance_uid: - raise ValueError( - f"No instance for step {step_num} — need CREATE before non-create operations") - display_name = None - descr = None - - # ═══ Шаг 4: запуск через ЕДИНЫЙ executor ═══ - # Это гарантирует идентичное поведение с ручным режимом - result = execute_operation( - client, svc_id, op_name, instance_uid, resolved_params, - svc_op_id=svc_op_id, display_name=display_name, descr=descr, - client_id=client_id, stand=stand - ) - if not result["ok"]: - raise RuntimeError( - f"{op_name}: {result['error']} (step: {result['failed_step']})") - - instance_uid = result["instance_uid"] - op_uid = result["op_uid"] - step_display = result["display_name"] - - # Обновить карты для резолвинга следующих шагов - instance_map[svc_id] = instance_uid # fallback: последний инстанс сервиса - output_name = (step.get("output") or "").strip() - if is_create and output_name: - bindings[output_name] = instance_uid # именованная ссылка - _update_bindings(scenario_run_id, bindings) - - # ═══ Шаг 5: сохранить RUNNING в БД (до поллинга) ═══ try: - save_run(client_id, stand, user_email, - svc_id, "", op_name, svc_op_id, - op_uid, instance_uid, step_display, - "RUNNING", 0, "", - resolved_params, [], app_version, - scenario_run_id, step_num) - except Exception as e: - print(f"[SCENARIO] save_run(RUNNING) failed for step {step_num}: {e}", flush=True) + # Валидация базовых полей шага + if not isinstance(svc_id, int) or svc_id <= 0: + raise ValueError(f"Invalid service_id: {svc_id!r}") + if not op_name: + raise ValueError("Empty operation name") - # ═══ Шаг 6: поллинг до завершения ═══ - poll_result = poll_until_done(client, op_uid) - step_status = poll_result["status"] # OK / FAIL / TIMEOUT - step_error = poll_result["error_log"] - step_duration = poll_result["duration"] - step_stages = poll_result["stages"] - svc_final = poll_result["svc"] + # ═══ Шаг 1: получить svcOperationId для операции ═══ + detail = get_service_detail(client, svc_id) + ops = detail.get("operations", []) + op = next((o for o in ops if o.get("operation") == op_name), None) + if not op: + raise ValueError(f"Operation '{op_name}' not found for service {svc_id}") + svc_op_id = op["svcOperationId"] - # ═══ Шаг 7: сохранить финальный результат + instance_meta ═══ - try: - # Получить полную информацию об инстансе для анализа - instance_meta = None + # ═══ Шаг 2: резолвить параметры (символ. имена → числовые ID) ═══ + tmpl_data = client.get(f"/instanceOperations/default/{svc_op_id}") + cfs_params = tmpl_data.get("svcOperation", {}).get("cfsParams", []) + resolved_params = _resolve_params(cfs_params, symbolic_params) + + # ═══ Шаг 3: резолвинг instance_uid ═══ + is_create = (op_name == "create") + if is_create: + # Для create: instance_uid=None (создаём новый) + # displayName генерируем уникальный: autotest-scenario-{name}-{random6} + instance_uid = None + display_name = f"{AUTOTEST_PREFIX}{scenario_name}-{uuid.uuid4().hex[:6]}" + descr = f"scenario {scenario_name} step {step_num}" + else: + # Для не-create: резолвим instance_uid + instance_uid = _resolve_instance_uid(step, bindings, instance_map) + if not instance_uid: + raise ValueError( + f"No instance for step {step_num} — need CREATE before non-create operations") + display_name = None + descr = None + + # ═══ Шаг 4: запуск через ЕДИНЫЙ executor ═══ + # Это гарантирует идентичное поведение с ручным режимом + result = execute_operation( + client, svc_id, op_name, instance_uid, resolved_params, + svc_op_id=svc_op_id, display_name=display_name, descr=descr, + client_id=client_id, stand=stand + ) + if not result["ok"]: + raise RuntimeError( + f"{op_name}: {result['error']} (step: {result['failed_step']})") + + instance_uid = result["instance_uid"] + op_uid = result["op_uid"] + step_display = result["display_name"] + + # Обновить карты для резолвинга следующих шагов + instance_map[svc_id] = instance_uid # fallback: последний инстанс сервиса + output_name = (step.get("output") or "").strip() + if is_create and output_name: + bindings[output_name] = instance_uid # именованная ссылка + _update_bindings(scenario_run_id, bindings) + + # ═══ Шаг 5: сохранить RUNNING в БД (до поллинга) ═══ try: - inst_data = client.get(f"/instances/{instance_uid}") - inst = inst_data.get("instance", {}) if isinstance(inst_data, dict) else {} - if inst: - # Сохраняем ВСЁ кроме очевидных/избыточных полей - instance_meta = {k: v for k, v in inst.items() - if k not in ('instanceUid', 'displayName', 'svc', 'serviceId', 'explainedStatus')} - except Exception: - pass # не смогли получить meta — не критично - save_run(client_id, stand, user_email, - svc_id, svc_final, op_name, svc_op_id, - op_uid, instance_uid, step_display, - step_status, step_duration, step_error, - resolved_params, step_stages, app_version, - scenario_run_id, step_num, - instance_meta=instance_meta) + save_run(client_id, stand, user_email, + svc_id, "", op_name, svc_op_id, + op_uid, instance_uid, step_display, + "RUNNING", 0, "", + resolved_params, [], app_version, + scenario_run_id, step_num) + except Exception as e: + print(f"[SCENARIO] save_run(RUNNING) failed for step {step_num}: {e}", flush=True) + + # ═══ Шаг 6: поллинг до завершения ═══ + poll_result = poll_until_done(client, op_uid) + step_status = poll_result["status"] # OK / FAIL / TIMEOUT + step_error = poll_result["error_log"] + step_duration = poll_result["duration"] + step_stages = poll_result["stages"] + svc_final = poll_result["svc"] + + # ═══ Шаг 7: сохранить финальный результат + instance_meta ═══ + try: + # Получить полную информацию об инстансе для анализа + instance_meta = None + try: + inst_data = client.get(f"/instances/{instance_uid}") + inst = inst_data.get("instance", {}) if isinstance(inst_data, dict) else {} + if inst: + # Сохраняем ВСЁ кроме очевидных/избыточных полей + instance_meta = {k: v for k, v in inst.items() + if k not in ('instanceUid', 'displayName', 'svc', 'serviceId', 'explainedStatus')} + except Exception: + pass # не смогли получить meta — не критично + save_run(client_id, stand, user_email, + svc_id, svc_final, op_name, svc_op_id, + op_uid, instance_uid, step_display, + step_status, step_duration, step_error, + resolved_params, step_stages, app_version, + scenario_run_id, step_num, + instance_meta=instance_meta) + except Exception as e: + print(f"[SCENARIO] save_run failed for step {step_num}: {e}", flush=True) + + # ═══ Шаг 8: обновить scenario_runs (прогресс для UI) ═══ + # Пока не последний шаг — статус RUNNING + _save_scenario_run(scenario_run_id, + "RUNNING" if step_num < total else step_status, + step_num, round(time.time() - t0, 1), + step_error if step_status != "OK" else None) + + # Если шаг FAIL — останавливаем сценарий + if step_status != "OK": + _save_scenario_run(scenario_run_id, step_status, step_num, + round(time.time() - t0, 1), step_error) + return + except Exception as e: - print(f"[SCENARIO] save_run failed for step {step_num}: {e}", flush=True) - - # ═══ Шаг 8: обновить scenario_runs (прогресс для UI) ═══ - # Пока не последний шаг — статус RUNNING - _save_scenario_run(scenario_run_id, - "RUNNING" if step_num < total else step_status, - step_num, round(time.time() - t0, 1), - step_error if step_status != "OK" else None) - - # Если шаг FAIL — останавливаем сценарий - if step_status != "OK": - _save_scenario_run(scenario_run_id, step_status, step_num, - round(time.time() - t0, 1), step_error) + # Любая ошибка → FAIL, сценарий остановлен + import traceback + err = f"Step {step_num}: {e}" + print(f"[SCENARIO] {err}", flush=True) + traceback.print_exc() + _save_scenario_run(scenario_run_id, "FAIL", step_num, + round(time.time() - t0, 1), err) return - except Exception as e: - # Любая ошибка → FAIL, сценарий остановлен - import traceback - err = f"Step {step_num}: {e}" - print(f"[SCENARIO] {err}", flush=True) - traceback.print_exc() - _save_scenario_run(scenario_run_id, "FAIL", step_num, - round(time.time() - t0, 1), err) - return - - # Все шаги пройдены успешно - _save_scenario_run(scenario_run_id, "OK", total, round(time.time() - t0, 1)) + # Все шаги пройдены успешно + _save_scenario_run(scenario_run_id, "OK", total, round(time.time() - t0, 1)) + finally: + # Освободить advisory lock ВСЕГДА (даже при исключении) + unlock_scenario(client_id, stand) diff --git a/site/operations/terraform.py b/site/operations/terraform.py index 73a2524..e15f0f6 100644 --- a/site/operations/terraform.py +++ b/site/operations/terraform.py @@ -245,13 +245,11 @@ def send_params_terraform(client, op_uid, params): # Шаг 7: validate-cfs — финальная проверка # Если параметры невалидны — API вернёт ошибку → исключение - # Если ответ пустой или не-JSON — это ОК (значит валидация прошла) + # Если ответ 200 с пустым телом — это ОК (значит валидация прошла). + # HttpClient.get() для пустого тела возвращает {} без ошибок. + # Но если тело не-JSON — будет JSONDecodeError, это тоже ОК для validate-cfs. try: client.get(f"/instanceOperations/{op_uid}/validate-cfs") - except Exception as e: - # "Expecting value" / JSONDecodeError — это норма для validate-cfs - # (API возвращает 200 с пустым телом при успехе) - if "Expecting value" in str(e) or "JSON" in str(type(e).__name__): - pass - else: - raise # реальная ошибка — пробрасываем выше + except json.JSONDecodeError: + pass # пустой/не-JSON ответ — норма для validate-cfs (успех) + # HTTPError (4xx/5xx) пробрасывается выше — это реальная ошибка валидации diff --git a/site/routes/api_test.py b/site/routes/api_test.py index 6237eef..4847f3b 100644 --- a/site/routes/api_test.py +++ b/site/routes/api_test.py @@ -24,6 +24,7 @@ """ from flask import Blueprint, current_app, jsonify, request +import threading import uuid import os import json @@ -261,23 +262,27 @@ def api_test(): # Результаты фоновых операций: opUid → {status, error, stages, duration, _ts} # _ts — timestamp добавления, для TTL-очистки (макс. 500 записей или старше 1 часа) +# ЗАЩИЩЕНО _op_results_lock — несколько потоков _finish_op + main thread api_test_status _op_results = {} +_op_results_lock = threading.Lock() _MAX_OP_RESULTS = 500 def _cleanup_op_results(): - """Удалить старые записи: старше 1 часа или сверх лимита.""" + """Удалить старые записи: старше 1 часа или сверх лимита. + Потокобезопасно — под _op_results_lock.""" import time now = time.time() - # Удалить старше 1 часа - stale = [k for k, v in _op_results.items() if now - v.get("_ts", 0) > 3600] - for k in stale: - del _op_results[k] - # Если всё ещё много — удалить самые старые - if len(_op_results) > _MAX_OP_RESULTS: - sorted_keys = sorted(_op_results.keys(), key=lambda k: _op_results[k].get("_ts", 0)) - for k in sorted_keys[:len(_op_results) - _MAX_OP_RESULTS]: - del _op_results[k] + with _op_results_lock: + # Удалить старше 1 часа (pop с default — безопасно при конкурентном доступе) + stale = [k for k, v in _op_results.items() if now - v.get("_ts", 0) > 3600] + for k in stale: + _op_results.pop(k, None) + # Если всё ещё много — удалить самые старые + if len(_op_results) > _MAX_OP_RESULTS: + sorted_keys = sorted(_op_results.keys(), key=lambda k: _op_results[k].get("_ts", 0)) + for k in sorted_keys[:len(_op_results) - _MAX_OP_RESULTS]: + _op_results.pop(k, None) def _get_instance_display_name(client, instance_uid): @@ -291,7 +296,7 @@ def _get_instance_display_name(client, instance_uid): def _finish_op(client, op_uid, instance_uid, svc_id, display_name, op_name, svc_op_id, is_create, client_id, stand, is_delete=False, params=None, user_email="", app_version=""): - """Фоном ждать dtFinish и сохранить результат.""" + """Фоном ждать dtFinish и сохранить результат (потокобезопасно для _op_results).""" import time _cleanup_op_results() t0 = time.time() @@ -299,15 +304,16 @@ def _finish_op(client, op_uid, instance_uid, svc_id, display_name, op_name, svc_ # Поллинг через общий модуль poll_result = poll_until_done(client, op_uid) - # Обновить _op_results для UI - _op_results[op_uid] = { - "status": poll_result["status"], - "displayName": display_name, - "error": poll_result["error_log"], - "stages": poll_result["stages"], - "duration": poll_result["duration"], - "_ts": time.time(), - } + # Обновить _op_results для UI — под локом + with _op_results_lock: + _op_results[op_uid] = { + "status": poll_result["status"], + "displayName": display_name, + "error": poll_result["error_log"], + "stages": poll_result["stages"], + "duration": poll_result["duration"], + "_ts": time.time(), + } # tracker_remove при успешном delete if poll_result["status"] == "OK" and is_delete: @@ -346,10 +352,11 @@ def _finish_op(client, op_uid, instance_uid, svc_id, display_name, op_name, svc_ @bp.route("/api/test/status/") def api_test_status(op_uid): - """Получить текущий статус операции (поллинг с UI).""" - # сначала проверяем фоновый трекер - if op_uid in _op_results: - return jsonify(_op_results[op_uid]) + """Получить текущий статус операции (поллинг с UI) — потокобезопасно.""" + # сначала проверяем фоновый трекер — под локом + with _op_results_lock: + if op_uid in _op_results: + return jsonify(_op_results[op_uid]) # иначе спрашиваем API напрямую try: data = get_client().get(f"/instanceOperations/{op_uid}?fields=dtFinish,isSuccessful,errorLog,isInProgress,duration,stages") diff --git a/site/static/js/scenario-form.js b/site/static/js/scenario-form.js index 8c79c0d..1df8a77 100644 --- a/site/static/js/scenario-form.js +++ b/site/static/js/scenario-form.js @@ -208,6 +208,9 @@ async function loadStepParams(idx) { const step = st.steps[idx]; if (!step.operation) return; + // Запомнить поколение рендера на момент старта async-загрузки + const renderGen = st._renderGen || 0; + const svcOpId = await resolveStepSvcOpId(idx); if (!svcOpId) return; step._svcOpId = svcOpId; // сохраняем для обратного маппинга при сохранении @@ -233,6 +236,8 @@ async function loadStepParams(idx) { // Рендер: клонируем def и подставляем сохранённые значения в defaultValue const container = document.getElementById('step-' + idx + '-params'); if (!container) return; + // Проверка: не перезаписывать DOM если уже был новый renderEditor() + if (st._renderGen !== renderGen) return; let html = ''; defs.forEach(p => { const clone = Object.assign({}, p); // shallow copy @@ -259,7 +264,10 @@ async function loadStepParams(idx) { function renderEditor() { snapshotAllParams(); // сохранить несохранённые правки перед перерисовкой const st = scenarioEditorState; - + // Инкремент поколения — защита от stale async (loadStepParams может вернуться + // уже после следующего renderEditor и перезаписать новый DOM старыми данными) + st._renderGen = (st._renderGen || 0) + 1; + const currentGen = st._renderGen; let html = '
'; // Заголовок