FastAPI Utils repeat_every在Cloud Run运行数小时后失效问题排查

我有一个每15分钟运行一次的定时任务，代码如下：

@app.on_event("startup")
@repeat_every(seconds=60 * 15, wait_first=True)
def myFunction(db=SessionLocal()) -> None:
    test(db=db, for_test=False)

该任务初始运行正常，但运行5-6小时后便停止重复执行。查看Cloud Run日志时，发现以下错误信息：

RuntimeError: coroutine ignored GeneratorExit 

同时还有如下日志：

2022-09-29 10:04:04.263 EETTask was destroyed but it is pending!
2022-09-29 10:04:04.263 EETtask: <Task pending name='Task-850' coro=<H2Protocol.send_task() running at /usr/local/lib/python3.9/site-packages/hypercorn/protocol/h2.py:148> wait_for=<Future cancelled> cb=[_gather.<locals>._done_callback() at /usr/local/lib/python3.9/asyncio/tasks.py:767]>

必须部署新版本才能恢复定时任务运行。想了解该错误的原因、解决办法，以及是否应该改用Google Scheduler替代FastAPI Utils的定时任务。

• Cloud Run实例伸缩机制：Cloud Run会在实例长时间空闲、资源超限或系统维护时自动重启/回收实例。你的定时任务依赖FastAPI应用进程存活，一旦实例被终止，任务就会中断。
• FastAPI Utils任务的进程绑定性：repeat_every装饰器创建的是应用进程内的后台任务，进程终止时，未完成的协程会被强制销毁，日志中的GeneratorExit和未完成任务提示，就是进程被终止时的典型表现。
• 同步代码阻塞事件循环：如果test是同步函数，直接在repeat_every的任务中执行会阻塞异步事件循环，导致协程无法正常调度，长期积累可能触发进程异常终止。

针对FastAPI Utils任务的临时修复 #

• 异步兼容改造：将同步任务包装为异步执行，避免阻塞事件循环：

import asyncio
from fastapi_utils.tasks import repeat_every

@app.on_event("startup")
@repeat_every(seconds=60 * 15, wait_first=True)
async def myFunction():
    db = SessionLocal()
    try:
        # 用线程池执行同步test函数
        await asyncio.to_thread(test, db=db, for_test=False)
    finally:
        db.close()

• 添加全局异常捕获：避免单个任务失败导致整个定时循环终止：

@app.on_event("startup")
@repeat_every(seconds=60 * 15, wait_first=True)
async def myFunction():
    try:
        db = SessionLocal()
        await asyncio.to_thread(test, db=db, for_test=False)
    except Exception as e:
        print(f"定时任务执行失败: {str(e)}")
    finally:
        db.close()

• 配置Cloud Run最小实例数：部署时设置--min-instances=1，强制保持至少一个实例运行，避免缩容导致进程终止，但会增加持续运行的成本。

长期可靠方案：改用Google Cloud Scheduler #

• 核心优势：Cloud Scheduler是独立的托管定时任务服务，不依赖FastAPI实例存活，通过HTTP请求触发Cloud Run接口，完全规避实例伸缩带来的任务中断问题，可靠性更高。
• 实现步骤：
  1. 在FastAPI中添加专用任务接口：

  from fastapi import Depends
  from sqlalchemy.orm import Session
  
  @app.post("/run-scheduled-task")
  def run_scheduled_task(db: Session = Depends(get_db)):
      test(db=db, for_test=False)
      return {"status": "success"}
  

  2. 在Google Cloud Console创建Scheduler任务：设置每15分钟触发一次，目标类型选HTTP，指向Cloud Run服务的/run-scheduled-task接口，配置服务账号权限确保能调用该接口。
  3. 移除原FastAPI中的repeat_every任务代码，任务逻辑完全由接口承接。

如果业务对定时任务的可靠性要求高，优先改用Google Cloud Scheduler，这是最适配Cloud Run环境的方案。若暂时无法切换，可先通过异步改造、异常捕获和最小实例配置临时解决，但成本和稳定性不如专用定时任务服务。

内容的提问来源于stack exchange，提问作者muzak