修复：定时任务

server/app/tasks/push.py

@@ -9,7 +9,7 @@ from zoneinfo import ZoneInfo
 
 import httpx
 from celery import current_app, shared_task
-from sqlalchemy import select
+from sqlalchemy import select, update
 
 from app.core.config import get_settings
 from app.db.models.push_preference import PushPreference
@@ -167,16 +167,25 @@ async def _generate_schedule_once(*, now_utc: datetime, max_users: int = 5000) -
                 created += 1
 
                 # 投递 ETA 发送任务
-                current_app.send_task(
-                    "tasks.push.send_scheduled",
-                    kwargs={
-                        "client_user_id": pref.client_user_id,
-                        "local_date": target.local_date.isoformat(),
-                        "slot_index": int(idx),
-                    },
-                    eta=dt_utc,
-                )
-                scheduled += 1
+                try:
+                    current_app.send_task(
+                        "tasks.push.send_scheduled",
+                        kwargs={
+                            "client_user_id": pref.client_user_id,
+                            "local_date": target.local_date.isoformat(),
+                            "slot_index": int(idx),
+                        },
+                        eta=dt_utc,
+                    )
+                    scheduled += 1
+                except Exception as e:
+                    # 关键：如果投递失败（例如 broker 短暂不可用），不要让 log 永远卡在 scheduled
+                    log.status = "failed"
+                    log.error = f"enqueue_failed:{type(e).__name__}"
+                    try:
+                        await session.commit()
+                    except Exception:
+                        await session.rollback()
 
     return {"created": created, "scheduled": scheduled}
 
@@ -209,6 +218,24 @@ async def _send_once_async(*, client_user_id: str, local_date: date, slot_index:
             return {"status": "noop", "reason": "no_log"}
         if str(log.status) == "sent":
             return {"status": "noop", "reason": "already_sent"}
+        if str(log.status) != "scheduled":
+            # 例如 failed/skipped/sending：不再重复尝试
+            return {"status": "noop", "reason": f"not_scheduled:{log.status}"}
+
+        # 原子抢占：避免同一条 scheduled 被重复发送（例如 ETA 任务丢失后被补偿重投递）
+        # 只有从 scheduled -> sending 抢占成功的任务才能继续执行。
+        res = await session.execute(
+            update(PushSendLog)
+            .where(
+                PushSendLog.id == log.id,
+                PushSendLog.status == "scheduled",
+            )
+            .values(status="sending", error=None)
+        )
+        await session.commit()
+        if (res.rowcount or 0) <= 0:
+            return {"status": "noop", "reason": "already_in_progress_or_processed"}
+        log.status = "sending"
 
         # 2) 当前偏好检查（用户可能中途关闭/改次数）
         qpref = select(PushPreference).where(PushPreference.client_user_id == client_user_id)
@@ -310,3 +337,46 @@ def send_scheduled(*, client_user_id: str, local_date: str, slot_index: int) ->
     d = date.fromisoformat(str(local_date))
     return asyncio.run(_send_once_async(client_user_id=str(client_user_id), local_date=d, slot_index=int(slot_index)))
 
+
+@shared_task(name="tasks.push.requeue_overdue")
+def requeue_overdue(*, grace_seconds: int = 300, limit: int = 200) -> dict[str, Any]:
+    """
+    补偿任务：扫描“已到时间但仍处于 scheduled”的记录并重新投递发送任务。
+
+    目的：
+    - 覆盖 broker 短暂不可用、worker 重启、ETA 任务丢失等导致的“scheduled 卡住”
+    - 与 send_scheduled 内部的原子状态抢占配合，避免重复发送
+    """
+
+    now_utc_naive = datetime.now(timezone.utc).replace(tzinfo=None)
+    cutoff = now_utc_naive - timedelta(seconds=int(grace_seconds))
+
+    async def _run() -> dict[str, Any]:
+        requeued = 0
+        async with AsyncSessionLocal() as session:
+            q = (
+                select(PushSendLog)
+                .where(PushSendLog.status == "scheduled", PushSendLog.scheduled_at <= cutoff)
+                .order_by(PushSendLog.scheduled_at.asc())
+                .limit(int(limit))
+            )
+            rows = await session.execute(q)
+            logs = list(rows.scalars().all())
+            for log in logs:
+                try:
+                    current_app.send_task(
+                        "tasks.push.send_scheduled",
+                        kwargs={
+                            "client_user_id": str(log.client_user_id),
+                            "local_date": str(log.local_date),
+                            "slot_index": int(log.slot_index),
+                        },
+                    )
+                    requeued += 1
+                except Exception:
+                    # 忽略单条投递失败，交给下一轮补偿
+                    continue
+        return {"status": "ok", "requeued": requeued, "cutoff": cutoff.isoformat()}
+
+    return asyncio.run(_run())
+

server/app/worker.py

@@ -57,6 +57,14 @@ celery_app.conf.beat_schedule = {
         "kwargs": {"max_users": 5000},
         "options": {"queue": f"{prefix}:celery"},
     }
+    ,
+    # 补偿：每 5 分钟扫描一次 overdue scheduled 并重投递
+    "push-requeue-overdue": {
+        "task": "tasks.push.requeue_overdue",
+        "schedule": crontab(minute="*/5"),
+        "kwargs": {"grace_seconds": 300, "limit": 200},
+        "options": {"queue": f"{prefix}:celery"},
+    },
 }
 
 # 自动发现任务（约定：导入 app.tasks 触发其内部对子模块的显式导入）