From e4cf7799a43cd50424260bc31ad9048918af66c2 Mon Sep 17 00:00:00 2001 From: GinHa <15201596918@163.com> Date: Fri, 3 Jul 2026 12:31:01 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8Dpoll=E9=99=8D=E9=A2=91?= =?UTF-8?q?=E6=A3=80=E6=B5=8B=E6=97=B6=E9=97=B4BUG?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../services/generation_recovery_service.py | 59 +++++++++++++++---- .../app/tasks/generation_poll_tasks.py | 11 ++-- 2 files changed, 55 insertions(+), 15 deletions(-) diff --git a/video-gen-api/app/services/generation_recovery_service.py b/video-gen-api/app/services/generation_recovery_service.py index bf455de3..4692032e 100644 --- a/video-gen-api/app/services/generation_recovery_service.py +++ b/video-gen-api/app/services/generation_recovery_service.py @@ -446,7 +446,12 @@ async def recover_one_generation_task( message=f"{source} 发现任务已到 deadline 且存在供应商任务ID，投递 poll 队列做最终查询", detail={"pipeline_stage": task.pipeline_stage, "payload": redis_payload}, ) - poll_generation_task.apply_async(args=[task.id], queue=POLL_QUEUE, countdown=0) + poll_generation_task.apply_async( + args=[task.id], + kwargs={"force_due": True}, + queue=POLL_QUEUE, + countdown=0, + ) await register_poll_active( task, check_at=_poll_queue_timeout_at(), @@ -487,16 +492,30 @@ async def recover_one_generation_task( ) return "skip_video_poll_not_due" + original_next_poll_at = ensure_aware_utc(task.next_poll_at) + queue_hold_until = _poll_queue_timeout_at(current_time) task.pipeline_stage = ChatGenerationPipelineStage.WAITING_REMOTE.value - task.next_poll_at = _poll_queue_timeout_at(current_time) + # 这里仍复用 next_poll_at 做短暂队列保护，避免启动容灾重复投递。 + # 真正消费时通过 force_due=True 跳过“未到期”校验，避免保护时间反向阻塞本次 poll。 + task.next_poll_at = queue_hold_until await db.commit() await log_task_event( task, event_type=ChatGenerationTaskEventType.GENERATION_RECOVERY_ENQUEUE.value, message=f"{source} 发现任务存在供应商任务ID，恢复投递轮询队列", - detail={"pipeline_stage": task.pipeline_stage, "payload": redis_payload}, + detail={ + "pipeline_stage": task.pipeline_stage, + "payload": redis_payload, + "due_next_poll_at": original_next_poll_at, + "queue_hold_until": queue_hold_until, + }, + ) + poll_generation_task.apply_async( + args=[task.id], + kwargs={"force_due": True}, + queue=POLL_QUEUE, + countdown=0, ) - poll_generation_task.apply_async(args=[task.id], queue=POLL_QUEUE, countdown=0) await register_poll_active( task, check_at=task.next_poll_at, @@ -696,6 +715,8 @@ async def dispatch_due_poll_tasks_once(db: AsyncSession) -> dict[str, Any]: results: dict[str, int] = {} dispatched_task_ids: list[str] = [] + dispatched_due_next_poll_at_by_id: dict[str, datetime | None] = {} + dispatched_queue_hold_until_by_id: dict[str, datetime] = {} for task in tasks: action = "skip_unknown" @@ -723,9 +744,14 @@ async def dispatch_due_poll_tasks_once(db: AsyncSession) -> dict[str, Any]: action = "skip_no_provider_task_id" continue + original_next_poll_at = ensure_aware_utc(task.next_poll_at) + queue_hold_until = _poll_queue_timeout_at(current_time) task.pipeline_stage = ChatGenerationPipelineStage.WAITING_REMOTE.value # 设置一个队列消费保护时间，避免 Beat 下一分钟看到旧 next_poll_at 又重复投递。 - task.next_poll_at = _poll_queue_timeout_at(current_time) + # poll worker 会通过 force_due=True 消费本次到期任务，避免该保护时间被误判为业务未到期。 + task.next_poll_at = queue_hold_until + dispatched_due_next_poll_at_by_id[task.id] = original_next_poll_at + dispatched_queue_hold_until_by_id[task.id] = queue_hold_until dispatched_task_ids.append(task.id) action = "dispatch_poll" except Exception as exc: @@ -745,30 +771,41 @@ async def dispatch_due_poll_tasks_once(db: AsyncSession) -> dict[str, Any]: if dispatched_task_ids: fresh_result = await db.execute( select(ChatGenerationTask) - .where( + .where( ChatGenerationTask.id.in_(dispatched_task_ids), ChatGenerationTask.deleted_at.is_(None), ) - .execution_options(populate_existing=True) + .execution_options(populate_existing=True) ) fresh_tasks = fresh_result.scalars().all() enqueued_count = 0 for task in fresh_tasks: + queue_hold_until = dispatched_queue_hold_until_by_id.get(task.id) or ensure_aware_utc(task.next_poll_at) or _poll_queue_timeout_at(current_time) + due_next_poll_at = dispatched_due_next_poll_at_by_id.get(task.id) await register_poll_active( task, - check_at=task.next_poll_at or _poll_queue_timeout_at(current_time), - next_poll_at=task.next_poll_at, + check_at=queue_hold_until, + next_poll_at=queue_hold_until, reason="due_dispatch_poll_queued", ) await log_task_event( task, event_type=ChatGenerationTaskEventType.POLL_DISPATCH_DUE.value, message="视频 next_poll_at 到期，已投递 provider poll 队列", - detail={"next_poll_at": task.next_poll_at, "queue": POLL_QUEUE}, + detail={ + "due_next_poll_at": due_next_poll_at, + "queue_hold_until": queue_hold_until, + "queue": POLL_QUEUE, + }, + ) + poll_generation_task.apply_async( + args=[task.id], + kwargs={"force_due": True}, + queue=POLL_QUEUE, + countdown=0, ) - poll_generation_task.apply_async(args=[task.id], queue=POLL_QUEUE, countdown=0) enqueued_count += 1 # 如果 log_task_event 内部不 commit，这里要提交一次 diff --git a/video-gen-api/app/tasks/generation_poll_tasks.py b/video-gen-api/app/tasks/generation_poll_tasks.py index 3ea1064c..31178aab 100644 --- a/video-gen-api/app/tasks/generation_poll_tasks.py +++ b/video-gen-api/app/tasks/generation_poll_tasks.py @@ -271,7 +271,7 @@ async def _schedule_next_poll( ) -async def _run(task_id: str): +async def _run(task_id: str, *, force_due: bool = False): async with async_session() as db: result = await db.execute(select(ChatGenerationTask).where( ChatGenerationTask.id == task_id, @@ -296,7 +296,10 @@ async def _run(task_id: str): current_time = _now() if is_video_generation_task(task): ensure_video_poll_fields(task, now=current_time) - if is_poll_not_due(task, now=current_time): + # dispatcher / recovery 已经在投递前确认到期时，会传 force_due=True。 + # 这样可以避免投递侧为了防重复消费临时写入的 next_poll_at， + # 又被当前 worker 当成“业务下一次轮询时间”而误判未到期。 + if not force_due and is_poll_not_due(task, now=current_time): await db.commit() await _skip_not_due(task) return @@ -481,9 +484,9 @@ async def _run(task_id: str): if celery_app: @celery_app.task(name="generation.poll_generation_task", bind=True, max_retries=3, default_retry_delay=30) - def poll_generation_task(self, task_id: str): + def poll_generation_task(self, task_id: str, force_due: bool = False): try: - return run_async(_run(task_id)) + return run_async(_run(task_id, force_due=bool(force_due))) except Exception as exc: # 只重试基础设施异常；供应商失败/业务失败已在 _run 内处理。 retries = int(getattr(self.request, "retries", 0) or 0) + 1