优化从 payment gateway 获取交易状态的重试间隔

结论

不要再按固定周期扫描所有 Pending 交易。应把“交易状态”和“下次查询时间”分开管理：

• 为每笔待确认交易保存 next_poll_at、attempt_count 等调度信息。
• 为 next_poll_at 建立索引，调度器只查询 next_poll_at <= now() 的到期任务。
• 多个 worker 使用租约或 FOR UPDATE SKIP LOCKED 并发领取任务。
• 查询后如果交易仍为 Pending，按照退避策略更新 next_poll_at。
• 优先使用 payment gateway 的 webhook 接收状态变化，轮询只负责补偿。
• 设置最大重试次数或最长等待时间，避免永久轮询。

cron 仍然可以保留，但只负责唤醒 worker，不再扫描整张表。

推荐的数据结构

如果交易表更新频繁，可以单独建立轮询任务表，避免把调度字段和核心交易数据混在一起。下面以 PostgreSQL 为例：

CREATE TABLE payment_poll_jobs (
    transaction_id  BIGINT PRIMARY KEY,
    attempt_count   INTEGER NOT NULL DEFAULT 0,
    next_poll_at    TIMESTAMPTZ NOT NULL,
    locked_until    TIMESTAMPTZ,
    locked_by       VARCHAR(100),
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX idx_payment_poll_jobs_due
    ON payment_poll_jobs (next_poll_at)
    WHERE next_poll_at IS NOT NULL;

交易本身仍保存在业务表中：

CREATE TABLE transactions (
    id                  BIGINT PRIMARY KEY,
    gateway_reference   VARCHAR(255) NOT NULL UNIQUE,
    status              VARCHAR(32) NOT NULL,
    status_updated_at   TIMESTAMPTZ NOT NULL DEFAULT now()
);

当 payment gateway 没有立即返回终态时，在同一个数据库事务中将交易标记为 Pending，同时创建轮询任务：

BEGIN;

UPDATE transactions
SET status = 'Pending',
    status_updated_at = now()
WHERE id = $1;

INSERT INTO payment_poll_jobs (
    transaction_id,
    attempt_count,
    next_poll_at
)
VALUES ($1, 0, now() + interval '1 minute')
ON CONFLICT (transaction_id) DO NOTHING;

COMMIT;

这里假设 X = 1 minute。实际的初始间隔需要根据 gateway 的正常处理时间、限流规则，以及业务对支付结果的时效要求来确定。

只领取已经到期的任务

调度器可以每几秒或每分钟运行一次。查询会使用 next_poll_at 索引，因此较高的调度频率并不等于频繁查询每笔交易。

多个 worker 并发运行时，可以通过租约领取任务：

WITH due_jobs AS (
    SELECT transaction_id
    FROM payment_poll_jobs
    WHERE next_poll_at <= now()
      AND (locked_until IS NULL OR locked_until < now())
    ORDER BY next_poll_at
    LIMIT 100
    FOR UPDATE SKIP LOCKED
)
UPDATE payment_poll_jobs AS job
SET locked_until = now() + interval '2 minutes',
    locked_by = $1,
    updated_at = now()
FROM due_jobs
WHERE job.transaction_id = due_jobs.transaction_id
RETURNING job.*;

这段语句应放在短事务中执行，领取完成后立即提交。不要在持有数据库行锁期间调用 payment gateway，否则外部接口一旦变慢，就会产生长事务和锁竞争。

locked_until 表示任务租约。worker 崩溃后，其他 worker 可以在租约过期时重新领取任务。租约时长应超过一次 gateway 请求的正常超时时间，但不宜设置得太长。

计算下一次重试时间

如果业务明确要求采用 X、5X、15X 这样的间隔，可以直接配置一组固定值：

const retryDelaysMs = [
  1 * 60_000,      // X
  5 * 60_000,      // 5X
  15 * 60_000,     // 15X
  30 * 60_000,
  60 * 60_000,
  3 * 60 * 60_000
];

function getRetryDelay(attemptCount: number): number | null {
  return retryDelaysMs[attemptCount] ?? null;
}

更通用的方式是采用指数退避，并限制最大间隔：

function getRetryDelay(
  attemptCount: number,
  baseDelayMs = 60_000,
  maxDelayMs = 6 * 60 * 60_000
): number {
  const delay = Math.min(
    baseDelayMs * 2 ** attemptCount,
    maxDelayMs
  );

  // 增加 ±20% jitter，避免大量任务在同一时刻访问 gateway。
  const jitterFactor = 0.8 + Math.random() * 0.4;
  return Math.floor(delay * jitterFactor);
}

退避策略不必限定为某个公式，但需要满足以下几点：

• 前几次查询的间隔较短，便于及时取得结果。
• 随着重试次数增加，逐步降低查询频率。
• 设置最大间隔，避免等待时间无限增加。
• 加入 jitter，防止大批交易同时发起请求。

Worker 的处理流程

下面是简化后的处理逻辑：

async function processPollJob(job: PaymentPollJob): Promise<void> {
  const transaction = await db.getTransaction(job.transaction_id);

  if (!transaction || transaction.status !== "Pending") {
    await db.deletePollJob(job.transaction_id);
    return;
  }

  let gatewayStatus: GatewayStatus;

  try {
    gatewayStatus = await gateway.getPaymentStatus(
      transaction.gateway_reference
    );
  } catch (error) {
    await reschedule(job, "gateway_request_failed");
    return;
  }

  if (
    gatewayStatus === "Succeeded" ||
    gatewayStatus === "Failed" ||
    gatewayStatus === "Cancelled"
  ) {
    await db.transaction(async (tx) => {
      await tx.updateTransactionStatus(
        transaction.id,
        gatewayStatus
      );
      await tx.deletePollJob(transaction.id);
    });

    return;
  }

  await reschedule(job, "still_pending");
}

async function reschedule(
  job: PaymentPollJob,
  reason: string
): Promise<void> {
  const delayMs = getRetryDelay(job.attempt_count);

  if (delayMs === null) {
    await db.markForManualReconciliation(
      job.transaction_id,
      reason
    );
    await db.deletePollJob(job.transaction_id);
    return;
  }

  await db.updatePollJob({
    transactionId: job.transaction_id,
    attemptCount: job.attempt_count + 1,
    nextPollAt: new Date(Date.now() + delayMs),
    lockedUntil: null,
    lockedBy: null
  });
}

在生产环境中，worker 完成任务或重新调度时，还要确认任务的租约仍归自己所有。较稳妥的做法是在领取任务时生成唯一的 lock_token，更新时带上它：

UPDATE payment_poll_jobs
SET attempt_count = attempt_count + 1,
    next_poll_at = $1,
    locked_until = NULL,
    locked_by = NULL,
    updated_at = now()
WHERE transaction_id = $2
  AND lock_token = $3;

这样可以防止租约过期后，旧 worker 覆盖新 worker 已写入的结果。

Webhook 应作为主要更新渠道

如果 payment gateway 支持 webhook，应主要通过 webhook 实时更新状态。轮询用于补偿以下情况：

• webhook 延迟；
• webhook 丢失；
• webhook 签名验证失败；
• 本地服务暂时不可用；
• gateway 与本地状态不一致。

Webhook 处理程序和 polling worker 可能同时更新同一笔交易，因此状态更新必须具备幂等性。可以根据 gateway 的事件 ID 去重，同时限制状态转换方向。例如，交易已经是 Succeeded 时，不应再被较旧的 Pending 事件覆盖。

常见的处理方式是：

1. 创建交易后，安排第一次短间隔轮询。
2. 收到有效 webhook 后，更新交易状态并删除轮询任务。
3. 没有收到 webhook 时，按照退避策略继续查询。
4. 达到最长等待时间后，将交易转入对账或人工处理。
5. 每天再运行一次低频 reconciliation，核对异常和遗漏数据。

是否需要消息队列

数据库调度可以满足很多系统的需求，尤其适合已有关系型数据库、任务规模中等，并且要求可靠持久化的场景。

任务量很大时，可以引入支持延迟消息的队列，但需要注意：

• 普通消息队列未必支持长时间或任意时间点的延迟任务。
• 消息可能重复投递，consumer 必须保证幂等。
• 不要只依靠队列保存长期支付状态，数据库仍是业务事实来源。
• 任务入队和数据库提交之间存在一致性问题，通常需要使用 transactional outbox。

建议先使用“数据库中的 next_poll_at + 索引 + worker 租约”。等数据库调度确实成为瓶颈后，再迁移到专门的任务调度系统或延迟队列。

需要特别注意的问题

• 为 gateway 请求设置连接超时和总超时，防止 worker 长时间占用租约。
• 根据 gateway 的 rate limit 控制全局并发，不能只控制单笔交易的重试间隔。
• 区分“仍在处理中”和“请求失败”。429、超时和 5xx 可以采用退避重试，鉴权失败等配置错误则不应无限重试。
• 设置 max_attempts 或 expires_at。支付状态超过合理期限仍无法确定时，应进入异常对账流程。
• 记录每次查询的时间、返回状态、错误类型和耗时，方便排查问题。
• gateway 返回终态后，在数据库事务中更新交易并移除任务。
• 通知、发货和记账等后续操作也要保证幂等，避免重复查询或重复 webhook 导致重复处理。

优化的重点不在于把 cron 变得更复杂，而是明确保存每笔交易的下次执行时间。调度器只处理已经到期的少量任务，再结合退避、jitter、租约、幂等和 webhook，才能兼顾及时性、可靠性和可扩展性。

备注：内容仅供参考。