fix: 兜底关联edit plan和回写config移到enqueue之前，消除竞态条件 (#1482)

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com> Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
2026-08-24 16:51:04 +08:00
parent 2827bf5943
commit fb4145f6de
2 changed files with 141 additions and 39 deletions
@@ -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: