From 066443b9c2e2db47312ee0a05a59e06dbe28624f Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 24 Aug 2026 16:30:22 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix:=20=E5=85=9C=E5=BA=95=E5=85=B3=E8=81=94?= =?UTF-8?q?edit=20plan=E5=92=8C=E5=9B=9E=E5=86=99config=E7=A7=BB=E5=88=B0e?= =?UTF-8?q?nqueue=E4=B9=8B=E5=89=8D=EF=BC=8C=E6=B6=88=E9=99=A4=E7=AB=9E?= =?UTF-8?q?=E6=80=81=E6=9D=A1=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因：safe_enqueue_generation_task之后才执行兜底关联和config回写， worker在enqueue后立即读取task时source_edit_plan_id仍为空。 修复： 1. 兜底关联edit plan移到enqueue之前执行 2. _writeback_edit_plan_config移到enqueue之前执行 3. 回写使用effective_plan_id（包含兜底结果），而非仅request.source_edit_plan_id 新增3个回归测试验证执行顺序正确。 --- apps/api/app/api/routes/generation_tasks.py | 82 +++++++------- .../unit/test_race_condition_fallback_plan.py | 106 ++++++++++++++++++ 2 files changed, 149 insertions(+), 39 deletions(-) create mode 100644 tests/unit/test_race_condition_fallback_plan.py diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index 9af9d0ecb..fe24672a2 100755 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/apps/api/app/api/routes/generation_tasks.py @@ -358,6 +358,49 @@ def create_generation_task( ) ) try: + # 兜底关联编辑计划：前端未传 source_edit_plan_id 时， + # 通过 template_id + user_id 在 DB 层直接查找最新的 plan。 + # 必须在 enqueue 之前执行，避免 worker 读取时 source_edit_plan_id 为空（竞态条件） + if not task.source_edit_plan_id and request.template_id: + try: + from packages.adapters.sqlalchemy_impl.models import EditPlanModel + + _plan_model = ( + db.query(EditPlanModel) + .filter( + EditPlanModel.template_id == request.template_id, + EditPlanModel.created_by_user_id == user_id, + ) + .order_by(EditPlanModel.created_at.desc()) + .first() + ) + if _plan_model: + task.source_edit_plan_id = _plan_model.id + generation_task_repository.update(task) + logger.info( + "[生成任务] 自动关联编辑计划: task_id=%s plan_id=%s", + task.id, + _plan_model.id, + ) + except Exception: + logger.warning( + "[生成任务] 查找关联编辑计划失败（不影响主流程）: task_id=%s", + task.id, + exc_info=True, + ) + + # 回写 plan.config：必须在 enqueue 之前执行， + # 确保 worker 读取 plan 时 config 中已包含 generation_task_id。 + # 只在首个任务时回写一次，避免批量生成时循环覆盖。 + _effective_plan_id = task.source_edit_plan_id + if _effective_plan_id and len(created_tasks) == 0: + _writeback_edit_plan_config( + plan_id=_effective_plan_id, + task_id=task.id, + title_config=request.title_config, + db=db, + ) + if safe_enqueue_generation_task( task, generation_task_repository, @@ -366,45 +409,6 @@ def create_generation_task( log_task_status=True, ): created_tasks.append(task) - # 只在首个成功任务时回写一次 plan.config， - # 避免批量生成时循环覆盖 generation_task_id - if request.source_edit_plan_id and len(created_tasks) == 1: - _writeback_edit_plan_config( - plan_id=request.source_edit_plan_id, - task_id=task.id, - title_config=request.title_config, - db=db, - ) - - # 兜底关联编辑计划：前端未传 source_edit_plan_id 时， - # 通过 template_id + user_id 在 DB 层直接查找最新的 plan - if not task.source_edit_plan_id and request.template_id: - try: - from packages.adapters.sqlalchemy_impl.models import EditPlanModel - - _plan_model = ( - db.query(EditPlanModel) - .filter( - EditPlanModel.template_id == request.template_id, - EditPlanModel.created_by_user_id == user_id, - ) - .order_by(EditPlanModel.created_at.desc()) - .first() - ) - if _plan_model: - task.source_edit_plan_id = _plan_model.id - generation_task_repository.update(task) - logger.info( - "[生成任务] 自动关联编辑计划: task_id=%s plan_id=%s", - task.id, - _plan_model.id, - ) - except Exception: - logger.warning( - "[生成任务] 查找关联编辑计划失败（不影响主流程）: task_id=%s", - task.id, - exc_info=True, - ) else: failed_tasks.append(task) except UserPendingLimitExceeded as _e: diff --git a/tests/unit/test_race_condition_fallback_plan.py b/tests/unit/test_race_condition_fallback_plan.py new file mode 100644 index 000000000..a527dde23 --- /dev/null +++ b/tests/unit/test_race_condition_fallback_plan.py @@ -0,0 +1,106 @@ +""" +回归测试：验证 generation_tasks.py 中兜底关联 edit plan 在 enqueue 之前执行。 + +根因（PR #1481 后续修复）：兜底关联逻辑原来在 safe_enqueue_generation_task 之后执行， +导致 worker 在 enqueue 后立即读取 task 时，source_edit_plan_id 仍为空（竞态条件）。 +""" + +import ast +import textwrap + + +def _get_function_source(filepath, func_name): + """提取函数源码""" + with open(filepath) as f: + source = f.read() + tree = ast.parse(source) + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == func_name: + lines = source.splitlines() + start = node.lineno - 1 + end = node.end_lineno + return textwrap.dedent("\n".join(lines[start:end])) + return None + + +def _find_try_block_source(func_source): + """在函数源码中找到包含 safe_enqueue_generation_task 的 try 块""" + tree = ast.parse(textwrap.dedent(func_source)) + for node in ast.walk(tree): + if isinstance(node, ast.Try): + # 检查 try 块中是否包含 safe_enqueue_generation_task + block_lines = func_source.splitlines() + block_text = "\n".join(block_lines[node.lineno - 1 : node.end_lineno]) + if "safe_enqueue_generation_task" in block_text: + return block_text + return None + + +def test_fallback_before_enqueue(): + """验证兜底关联 edit plan 的代码在 safe_enqueue_generation_task 调用之前""" + filepath = "apps/api/app/api/routes/generation_tasks.py" + func_source = _get_function_source(filepath, "create_generation_task") + assert func_source is not None, "create_generation_task function not found" + + try_block = _find_try_block_source(func_source) + assert try_block is not None, "try block with safe_enqueue_generation_task not found" + + # 定位关键标记在 try 块中的行号 + lines = try_block.splitlines() + + fallback_line = None + enqueue_line = None + writeback_line = None + + for i, line in enumerate(lines): + if "not task.source_edit_plan_id and request.template_id" in line and fallback_line is None: + fallback_line = i + if "safe_enqueue_generation_task(" in line and enqueue_line is None: + enqueue_line = i + if "_writeback_edit_plan_config(" in line and "def " not in line and writeback_line is None: + writeback_line = i + + assert fallback_line is not None, "兜底关联逻辑 not found in try block" + assert enqueue_line is not None, "safe_enqueue_generation_task call not found in try block" + assert writeback_line is not None, "_writeback_edit_plan_config call not found in try block" + + # 核心断言：兜底关联和回写都在 enqueue 之前 + assert fallback_line < enqueue_line, ( + f"兜底关联（行{fallback_line}）应在 enqueue（行{enqueue_line}）之前" + ) + assert writeback_line < enqueue_line, ( + f"回写 config（行{writeback_line}）应在 enqueue（行{enqueue_line}）之前" + ) + + +def test_fallback_sets_source_edit_plan_id(): + """验证兜底关联逻辑会设置 task.source_edit_plan_id""" + filepath = "apps/api/app/api/routes/generation_tasks.py" + func_source = _get_function_source(filepath, "create_generation_task") + assert func_source is not None + + try_block = _find_try_block_source(func_source) + assert try_block is not None + + # 验证兜底逻辑包含赋值语句 + assert "task.source_edit_plan_id = _plan_model.id" in try_block, ( + "兜底关联逻辑应设置 task.source_edit_plan_id" + ) + assert "generation_task_repository.update(task)" in try_block, ( + "兜底关联后应持久化 task 到 DB" + ) + + +def test_writeback_uses_effective_plan_id(): + """验证回写 config 使用的是 effective_plan_id（包含兜底结果），而非仅 request.source_edit_plan_id""" + filepath = "apps/api/app/api/routes/generation_tasks.py" + func_source = _get_function_source(filepath, "create_generation_task") + assert func_source is not None + + try_block = _find_try_block_source(func_source) + assert try_block is not None + + # 验证使用了 _effective_plan_id 或 task.source_edit_plan_id，而非仅 request.source_edit_plan_id + # 修复前用的是 request.source_edit_plan_id，修复后应该用 task.source_edit_plan_id + uses_effective = "_effective_plan_id" in try_block or "task.source_edit_plan_id" in try_block + assert uses_effective, "回写 config 应使用包含兜底结果的有效 plan_id" -- 2.54.0 From 12e880646233d22803072229c51a4b8b4491b2a3 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 24 Aug 2026 08:34:18 +0000 Subject: [PATCH 2/2] style: auto-format with black + isort + prettier [skip ci-format-check] --- tests/unit/test_race_condition_fallback_plan.py | 16 ++++------------ 1 file changed, 4 insertions(+), 12 deletions(-) diff --git a/tests/unit/test_race_condition_fallback_plan.py b/tests/unit/test_race_condition_fallback_plan.py index a527dde23..5690f22f5 100644 --- a/tests/unit/test_race_condition_fallback_plan.py +++ b/tests/unit/test_race_condition_fallback_plan.py @@ -65,12 +65,8 @@ def test_fallback_before_enqueue(): assert writeback_line is not None, "_writeback_edit_plan_config call not found in try block" # 核心断言：兜底关联和回写都在 enqueue 之前 - assert fallback_line < enqueue_line, ( - f"兜底关联（行{fallback_line}）应在 enqueue（行{enqueue_line}）之前" - ) - assert writeback_line < enqueue_line, ( - f"回写 config（行{writeback_line}）应在 enqueue（行{enqueue_line}）之前" - ) + assert fallback_line < enqueue_line, f"兜底关联（行{fallback_line}）应在 enqueue（行{enqueue_line}）之前" + assert writeback_line < enqueue_line, f"回写 config（行{writeback_line}）应在 enqueue（行{enqueue_line}）之前" def test_fallback_sets_source_edit_plan_id(): @@ -83,12 +79,8 @@ def test_fallback_sets_source_edit_plan_id(): assert try_block is not None # 验证兜底逻辑包含赋值语句 - assert "task.source_edit_plan_id = _plan_model.id" in try_block, ( - "兜底关联逻辑应设置 task.source_edit_plan_id" - ) - assert "generation_task_repository.update(task)" in try_block, ( - "兜底关联后应持久化 task 到 DB" - ) + assert "task.source_edit_plan_id = _plan_model.id" in try_block, "兜底关联逻辑应设置 task.source_edit_plan_id" + assert "generation_task_repository.update(task)" in try_block, "兜底关联后应持久化 task 到 DB" def test_writeback_uses_effective_plan_id(): -- 2.54.0