Celery Beat 事务邮件 + Rovela（2026）：crontab / beat_schedule 与按窗口幂等键

用 Celery Beat 周期任务（crontab 或间隔）驱动 worker 调用 Rovela POST /emails：sending_access 仅服务端，按调度窗口设 Idempotency-Key，适合摘要与运维报告。

Celery Beat 管调度，Rovela 负责投递

Celery Beat 是 Celery 的周期任务调度器。它从 beat_schedule（或 django-celery-beat 等自定义存储）读取条目，按 crontab、秒数 / timedelta 或自定义 schedule 判断是否到期，并把命名任务入队给 worker。Beat 本身不发邮件——真正执行的是 worker 里的任务代码。

这种形态适合摘要、运维报告、试用到期提醒等事务邮件：正文依赖触发时刻的实时应用 / 数据库状态（注册数、未付发票、失败任务），而不是可以提前几小时用 Rovela scheduled_at 冻住的内容。

当你的 Celery task 调用 POST https://api.rovela.dev/emails 时，Rovela 负责排队、DKIM 签名与投递（Send Email API、Quick Start、Authentication）。Celery Beat 拥有节奏与“是否到期”的逻辑；worker 任务拥有聚合、sending_access 密钥，以及按调度窗口稳定的 Idempotency-Key。Rovela 只负责投递。

本指南串联 Celery Beat → Celery task → Rovela，做到：

用 app.conf.beat_schedule / add_periodic_task 声明带 crontab（或间隔）的周期条目
把 ROVELA_API_KEY（sending_access）、MAIL_FROM、broker/backend URL、收件人放在仅服务端环境变量——绝不进浏览器或公开 Git
在 Celery task 内聚合当前调度窗口的摘要 / 报告
用按窗口作用域的 Idempotency-Key 调用 Rovela
在 Celery 重试、Beat 重启、重复 tick 下仍不重复发信

Rovela 不会创建 Celery Beat 计划、broker 或 worker。 Beat 管 schedule → 入队；你的 task 管查询 → 映射字段 → POST /emails；Rovela 管投递。

与下列文章区分：

Agenda.js 定时任务邮件 — Mongo 上的 agenda define / every（Node）
Temporal Schedule 邮件 — Temporal Schedule Spec + Activity
BullMQ 可重复 / Cron 任务邮件 — Redis BullMQ upsertJobScheduler
Trigger.dev Cron 邮件 — Trigger.dev schedules.task
Inngest Cron 事务邮件 — Inngest cron + step.run
Upstash QStash Schedules 邮件 — QStash HTTPS 调度 + Upstash-Signature
Supabase pg_cron 邮件 — Postgres pg_cron + pg_net
Kubernetes CronJob 邮件 — batch/v1 CronJob → Job Pod
DigitalOcean App Platform Jobs 邮件 — App Platform kind: SCHEDULED
Heroku Scheduler 邮件 — 粗粒度 Scheduler dyno
Cloudflare Workers Cron 邮件 — Workers triggers.crons
Azure Functions Timer 邮件 — Azure NCRONTAB
GCP Cloud Scheduler 邮件 — Cloud Scheduler → Cloud Functions / Pub/Sub
AWS EventBridge Scheduler 邮件 — EventBridge Scheduler → Lambda
Netlify Scheduled Functions 邮件 — Netlify UTC cron
Fly Scheduled Machines 邮件 — Fly Machine schedules
Render Cron 事务邮件 — Render Cron Job 进程
Railway Cron 事务邮件 — Railway cron 服务
Vercel Cron Digest 邮件 — 平台 cron GET 生产路由
GitHub Actions 事务邮件 — workflow / CI 通知
Scheduled Email Sending — 正文已知时用 Rovela scheduled_at

也不同于一次性的 Celery apply_async(countdown=…) / eta=——本文只写 Beat 周期任务路径（beat_schedule / crontab）。

Celery Beat 调度什么，Rovela 发送什么

关注点	归属
Beat 条目名、`task`、`schedule`（`crontab` / 秒 / `timedelta`）、`args` / `kwargs` / `options`	Celery `beat_schedule`
调度求值时区（默认 UTC）	Celery `timezone` / `app.conf.timezone`
上次运行持久化（`celerybeat-schedule` 或 DatabaseScheduler）	Beat 进程
到期时入队	Celery Beat（每个 schedule 只跑一个 Beat）
任务重试、确认、并发	Celery Worker + `autoretry_for` / `retry`
`ROVELA_API_KEY`（`sending_access`）密钥	worker 进程环境 / 密钥库
查询窗口内 DB / API	你的 Celery task
渲染 HTML + text	你的 Celery task
`POST /emails`、DKIM、排队、退信	Rovela
`Idempotency-Key` = `celery-beat-daily-digest:{yyyy-mm-dd}`（或小时 / 租户桶）	你的 Celery task

Rovela 返回 200 表示已接受并入队，不代表“已进收件箱”（Quick Start）。

影响发信任务的 Celery Beat 约束

依据 Celery Periodic Tasks 文档，发信调度请当作硬约束：

只跑一个 Beat。 文档写明同一 schedule 同时跑多个 scheduler 会导致重复任务。生产用独立的 celery -A proj beat，不要在每个 web dyno 上开 Beat。
生产勿用 worker -B。 把 Beat 嵌进 celery -A proj worker -B 适合单节点，不推荐生产（Starting the Scheduler）。
墙钟邮件用 crontab。 from celery.schedules import crontab 表达“周一 07:30”“每天午夜”“每 15 分钟”等。间隔调度用 float 秒或 timedelta（相对 Beat 启动，除非 relative=True）。
时区默认 UTC。 若 cron 要表示本地墙钟，设 app.conf.timezone = 'Asia/Tokyo' 等 IANA 区。PersistentScheduler 在时区变更时会重置；DatabaseScheduler 可能需手动清 last-run（Time Zones）。
可能重叠执行。 与 cron 一样，慢任务可能与下次 tick 重叠——若摘要只能单飞，加锁（Periodic Tasks）。
Beat 要写 last-run。 默认 PersistentScheduler 写 celerybeat-schedule（-s 可改路径），或用 django_celery_beat.schedulers:DatabaseScheduler。
Worker 必须在线。 Beat 只入队；没有消费队列的 worker，邮件不会发出。
sending_access 仅服务端。 ROVELA_API_KEY 放在 worker 环境，不要塞进可能被广泛记录的 task kwargs、客户端包或 Git（Authentication）。
优先 HTTPS POST /emails。 从 worker 直连 SMTP 通常更差（Email API vs SMTP）。

何时用 Celery Beat + `POST /emails`（而非 `scheduled_at`）

在这些情况下用 Celery Beat → 即时发送：

HTML 必须在触发时包含最新指标
已有 Celery worker（Redis/RabbitMQ），希望调度与同一应用相邻
需要类 cron 摘要，又不想另上 SaaS 调度器
一两封事务信发给运维 / 创始人 / 运行时查出的客户地址

当正文已经正确、只需要“稍后再投”且不必跑 Beat/worker 时，用 Rovela scheduled_at（Scheduled Email Sending）。Beat 与 scheduled_at 解决不同问题；本文是 Beat 路径。

若要用 BullMQ / Agenda / Temporal / Trigger.dev / Inngest / QStash，见上文区分链接；若调度应贴近某云主机算力，用对应平台 cron 文。

前置条件

Rovela 上已验证发信域名。未验证的 from 会 400（Quick Start）。
形如 re_… 的 API key。worker 优先 sending_access（Authentication）。
带 broker（Redis 或 RabbitMQ）的 Celery 应用，以及至少一个 worker。
worker / Beat 环境：ROVELA_API_KEY、MAIL_FROM、DIGEST_TO，以及 broker URL 与任务所需 DB URL。
该 schedule 恰好一个 Beat 进程（外加持续在线的 worker）。

声明 Beat 调度与任务（Python）

每天 09:00 UTC 的摘要，用 crontab：

# proj/celery.py
import os
from datetime import datetime, timezone

import httpx
from celery import Celery
from celery.schedules import crontab

app = Celery("proj")
app.config_from_object("django.conf:settings", namespace="CELERY")  # 或自有配置模块
# 非 Django 时：
# app.conf.broker_url = os.environ["CELERY_BROKER_URL"]
# app.conf.result_backend = os.environ.get("CELERY_RESULT_BACKEND")

app.conf.timezone = "UTC"

app.conf.beat_schedule = {
    "daily-digest-email": {
        "task": "proj.tasks.send_daily_digest_email",
        "schedule": crontab(minute=0, hour=9),  # 按 app.conf.timezone 每天 09:00
    },
}


def require_env(name: str) -> str:
    v = os.environ.get(name)
    if not v:
        raise ValueError(f"missing env {name}")
    return v


def utc_day_key(d: datetime | None = None) -> str:
    d = d or datetime.now(timezone.utc)
    return d.astimezone(timezone.utc).strftime("%Y-%m-%d")


def build_daily_digest(day: str) -> dict:
    # 换成你对 Postgres / Redis / API 的查询
    stats = {"signups": 12, "paid": 3, "incidents": 1, "failed_jobs": 0}
    subject = f"Acme daily digest — {day}"
    text = "\n".join(
        [
            f"Signups: {stats['signups']}",
            f"Paid: {stats['paid']}",
            f"Open incidents: {stats['incidents']}",
            f"Failed jobs: {stats['failed_jobs']}",
        ]
    )
    html = f"""
    <h1>Daily digest — {day}</h1>
    <ul>
      <li>Signups: <b>{stats['signups']}</b></li>
      <li>Paid: <b>{stats['paid']}</b></li>
      <li>Open incidents: <b>{stats['incidents']}</b></li>
      <li>Failed jobs: <b>{stats['failed_jobs']}</b></li>
    </ul>
    <p><a href="https://app.example.com/admin">Open admin</a></p>
    """
    return {"subject": subject, "html": html, "text": text, "stats": stats}


def send_transactional(
    *,
    to: list[str],
    subject: str,
    html: str,
    text: str,
    idempotency_key: str,
    tags: dict[str, str],
) -> dict:
    url = "https://api.rovela.dev/emails"
    headers = {
        "Authorization": f"Bearer {require_env('ROVELA_API_KEY')}",
        "Content-Type": "application/json",
        "Idempotency-Key": idempotency_key,
    }
    payload = {
        "from": require_env("MAIL_FROM"),
        "to": to,
        "subject": subject,
        "html": html,
        "text": text,
        "tags": tags,
    }
    with httpx.Client(timeout=30.0) as client:
        res = client.post(url, headers=headers, json=payload)
        body = res.json() if res.content else {}
        if res.status_code >= 400:
            raise RuntimeError(f"Rovela {res.status_code}: {body.get('message', 'send failed')}")
        return body


@app.task(
    name="proj.tasks.send_daily_digest_email",
    bind=True,
    autoretry_for=(httpx.TransportError, RuntimeError),
    retry_backoff=True,
    retry_kwargs={"max_retries": 5},
)
def send_daily_digest_email(self):
    day = utc_day_key()
    idempotency_key = f"celery-beat-daily-digest:{day}"

    digest = build_daily_digest(day)
    to = [s.strip() for s in require_env("DIGEST_TO").split(",") if s.strip()]

    # 切勿打印 ROVELA_API_KEY 或 broker 密码
    print(
        {
            "day": day,
            "idempotencyKey": idempotency_key,
            "taskId": self.request.id,
            "retries": self.request.retries,
        }
    )

    result = send_transactional(
        to=to,
        subject=digest["subject"],
        html=digest["html"],
        text=digest["text"],
        idempotency_key=idempotency_key,
        tags={
            "kind": "celery-beat-daily-digest",
            "day": day,
            "platform": "celery-beat",
            "task": self.name,
        },
    )
    return {"ok": True, "emailId": result.get("id"), "day": day}

也可用 add_periodic_task（避免模块级过早求值）：

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    sender.add_periodic_task(
        crontab(minute=0, hour=9),
        send_daily_digest_email.s(),
        name="daily-digest-email",
    )

生产中分开跑进程：

celery -A proj worker -l INFO
celery -A proj beat -l INFO -s /var/run/celerybeat-schedule
# 可选 Django DB scheduler：
# celery -A proj beat -l INFO --scheduler django_celery_beat.schedulers:DatabaseScheduler

本地时区早间运维信（先设 Beat 时区，再写该时区下的 crontab）：

app.conf.timezone = "Asia/Tokyo"

app.conf.beat_schedule = {
    "tokyo-morning-ops-email": {
        "task": "proj.tasks.send_tokyo_ops_email",
        "schedule": crontab(minute=0, hour=5),  # Asia/Tokyo 05:00
    },
}

# 任务内按调度时区的“日”做幂等键，例如 zoneinfo：
# from zoneinfo import ZoneInfo
# day = datetime.now(ZoneInfo("Asia/Tokyo")).strftime("%Y-%m-%d")
# idempotency_key = f"celery-beat-tokyo-ops:{day}"

固定间隔脉冲（秒 / timedelta）：

from datetime import timedelta

app.conf.beat_schedule = {
    "hourly-ops-email": {
        "task": "proj.tasks.send_hourly_ops_email",
        "schedule": 3600.0,  # 按 Beat 时钟每 3600 秒
        # 或: "schedule": timedelta(hours=1),
    },
}

# idempotency_key = f"celery-beat-hourly-ops:{utc_hour_key()}"
# utc_hour_key → YYYY-MM-DDTHH

适合发信节奏的 crontab 示例（文档表）：

Schedule	含义
`crontab(minute=0, hour=0)`	每天午夜（按 `timezone`）
`crontab(minute=0, hour=9)`	每天 09:00
`crontab(minute='*/15')`	每 15 分钟
`crontab(hour=7, minute=30, day_of_week=1)`	周一 07:30
`crontab(minute=0, hour='*/3')`	每三小时

按窗口作用域的 Idempotency-Key

Rovela 的 Idempotency-Key（最长 256 字符）让 24 小时内相同发送的重试返回原响应，而不是再发一封（Send Email API）。对 Celery 尤其重要：

任务 autoretry / retry 在 5xx / 超时后会重入
worker 在 Rovela 200 之后、ack 之前崩溃可能重投
误开两个 Beat（或另处还有 worker -B）会重复入队
队列积压会让 datetime.now() 漂移——键应按业务调度窗口（UTC 日 / 时区日 / 小时），不要用毫秒 now
多租户 kwargs 要把 tenant id 写进键

推荐键：

Key	何时用
`celery-beat-daily-digest:{yyyy-mm-dd}`	每个 UTC 日一封运维摘要
`celery-beat-hourly-ops:{yyyy-mm-dd}T{HH}`	小时间隔 / 小时 crontab
`celery-beat-tokyo-ops:{yyyy-mm-dd}`	本地时区早间摘要（按 Beat `timezone` 的日）
`celery-beat-weekly-report:{yyyy}-W{ww}`	每 ISO 周一封创始人报告
`celery-beat-tenant-digest:{tenantId}:{yyyy-mm-dd}`	按租户的 task kwargs

若一次任务要发很多个性化摘要，优先按收件人建键（或对最多 100 条非模板消息用 POST /emails/batch——batch 会拒绝 template / topic_id，属刻意设计（Send Email API））。

挂载密钥并验证调度

在 worker（以及若需要则 Beat）主机设置 ROVELA_API_KEY（sending_access）、MAIL_FROM、DIGEST_TO、broker URL。Beat 若只入队可不带 Rovela 密钥；worker 必须有。
部署一个 Beat + N 个 worker；确认 Beat 能写 -s / DB schedule。
确认 beat_schedule（或 django-celery-beat 后台）中 daily-digest-email 的 crontab / 时区正确。
等下次 tick，或在非生产用短间隔验证，同时保持窗口键稳定。
看 worker 日志中的 idempotencyKey + Rovela emailId；可选再用 webhook 关联。

生产发信优先 sending_access，不要用 full_access（Authentication）。切勿打印 API key。Rovela 轮换用先创建后吊销；broker 凭据按主机指引轮换。

故障模式与处理

现象	可能原因	处理
Beat 在跑但无执行	无 worker / 队列或路由错误	在正确队列起 worker；核对 `task` 名
摘要重复	两个 Beat / `worker -B` + Beat	只留一个 Beat；生产去掉嵌入式 `-B`
墙钟时间不对	`timezone` 仍是 UTC	设 `app.conf.timezone` 为你要的 IANA 区
摘要重叠	任务过长 + crontab过密	加锁 / 单飞；缩短工作
schedule 文件报错	Beat 写不了 `celerybeat-schedule`	修 `-s` 权限或改用 DatabaseScheduler
Rovela `401`	`ROVELA_API_KEY` 错误/缺失	修 worker 环境；确认 `sending_access`
Rovela `400` 提到域名	`MAIL_FROM` 域名未验证	在 Rovela Domains 验证
Rovela `429`	限流	抛错让 Celery 重试；保持同一 Idempotency-Key
重试后重复邮件	键不稳定 / 用了 `now()`	用窗口键；24h 内幂等键会折叠重复
密钥进 Git / kwargs	提交了密钥或塞进 kwargs	只放进程环境 / vault

生产清单

已验证 Rovela 发信域名；MAIL_FROM 匹配
ROVELA_API_KEY 为 sending_access，仅在 worker 环境
该 schedule 恰好一个 Celery Beat
已记录 crontab + timezone（或间隔）；任务重试已配置
对应队列至少有一个常驻 Worker
Rovela 发送使用按窗口的 Idempotency-Key
Beat schedule 存储可持久（-s 卷或 DatabaseScheduler）
日志有成功 emailId 且不打印密钥
对连续任务失败 / Beat 宕机告警
可选：对摘要收件人订阅 Rovela email.bounced / email.complained webhook

FAQ

Celery Beat 能不经 worker task 直接调 Rovela 吗？没有实用意义——Beat 只入队 Celery 任务。聚合实时数据与保管 sending_access 必须在你的 task 里。Rovela 不托管 Celery、Redis 或 RabbitMQ。

任务里该不该用 Rovela scheduled_at？通常不必。若 Beat 已在发送时刻带着实时正文触发，直接 POST /emails。当你现在收下消息、希望 Rovela 稍后再投且不必再醒一次时，才用 scheduled_at（Scheduled Email Sending）。

Celery Beat vs BullMQ vs Agenda vs Temporal vs Trigger.dev vs 平台 cron？调度应贴近 Python Celery worker 时用 Celery Beat。贴近 Node Redis/Mongo 队列时用 BullMQ / Agenda（BullMQ、Agenda）。要耐久 Schedule Spec 用 Temporal（Temporal）。要用那些 SaaS 运行时则见 Trigger.dev / Inngest / QStash。调度应贴近某云主机算力时用 Vercel/Render/K8s cron。

crontab 还是间隔秒数？墙钟摘要优先 crontab + timezone（本地 / UTC 的 09:00）。不需要对齐日历的固定脉冲用 秒 / timedelta。

celery worker -B 呢？笔记本或单节点演示可以。生产请单独跑 Beat，避免扩容 worker 时复制调度器（文档）。

sending_access 能做什么？足够通过 POST /emails / batch 入队发信。读消息、改期、取消需要 full_access 或仪表盘会话（Send Email API、Authentication）。worker 请保持 sending_access。

200 表示创始人已经收到摘要了吗？否——只表示 Rovela 已接受并入队。投递状态看 GET /emails/:id、events 或 webhook（Quick Start）。

为何既要 Celery 重试又要 Rovela Idempotency-Key？Celery 在代码失败时重试任务；Rovela 的键在同一窗口内 HTTP 成功两次（或超时后其实已接受）时折叠重复发送。两层都要。

上线 Beat 调度，投递交给 Rovela

用 crontab（或间隔）声明 Celery Beat 条目，把实时聚合与 POST /emails 放进 worker task，把 Idempotency-Key 限定在调度窗口（及租户），并让 sending_access 仅存服务端。Beat 按时唤醒；Rovela 投递事务邮件。