mz/video-gen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复poll降频检测时间BUG

Browse Source
This commit is contained in:
GinHa
2026-07-03 12:31:01 +08:00
parent 28b1d24173
commit e4cf7799a4
2 changed files with 55 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video-gen-api/app/services/generation_recovery_service.py
+48 -11
View File
@@ -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，这里要提交一次
video-gen-api/app/tasks/generation_poll_tasks.py
+7 -4
View File
@@ -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