fix(queue): 重新调度终态 River 任务

等待异步准入的任务只跳过仍处于活跃状态的 River Job;missing 或 terminal Job 交由准入调度器重新创建,避免非空 river_job_id 永久阻断旧任务。\n\n通用 River 恢复排除等待准入任务,消除双恢复循环与误导日志;事务回滚继续使用独立有界上下文。\n\n验证:真实 PostgreSQL 下异步 Worker 准入验收和跨 Store 队列集成测试均通过。
This commit is contained in:
2026-07-30 17:42:33 +08:00
parent e8893e2b8f
commit fa818e9ebf
5 changed files with 52 additions and 14 deletions
+9 -8
View File
@@ -502,7 +502,11 @@ func (s *Service) enqueueAsyncTaskWithOptions(ctx context.Context, taskID string
if err != nil {
return err
}
defer tx.Rollback(ctx)
defer func() {
rollbackCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_ = tx.Rollback(rollbackCtx)
}()
if err := s.enqueueAsyncTaskTx(ctx, tx, taskID, opts); err != nil {
return err
}
@@ -548,18 +552,15 @@ func (s *Service) recoverAsyncRiverJobs(ctx context.Context) error {
if err != nil {
return err
}
recovered := 0
for _, item := range items {
if admission, admissionErr := s.store.GetTaskAdmission(ctx, item.ID); admissionErr == nil && admission.Status == "waiting" {
continue
} else if admissionErr != nil && !errors.Is(admissionErr, pgx.ErrNoRows) {
return admissionErr
}
if err := s.enqueueAsyncTaskWithOptions(ctx, item.ID, asyncTaskRecoveryInsertOpts(item, time.Now())); err != nil {
return err
}
recovered++
}
if len(items) > 0 {
s.logger.Info("river async queue recovered persisted tasks", "count", len(items))
if recovered > 0 {
s.logger.Info("river async queue recovered persisted tasks", "count", recovered)
}
return nil
}