Slack同事件重复请求问题：已做x-slack-retry-num校验仍无效

背景 #

我们部署了两台服务器：backend-api（基于Django）和slack-api（基于FastAPI）。其中slack-api负责接收Slack UI的所有请求，读写数据库时调用backend-api。目前存在部分事件被Slack重复发送请求的情况，已经破坏了预期业务流程。

当前已在slack-api入口添加的校验逻辑：

retry_num = req.headers.get("x-slack-retry-num")
if retry_num is not None:
    return

但仍偶发同一事件收到多请求的问题，日志截图显示：同一事件ID对应多条请求记录，确认Slack重复发送了同一事件的请求。

问题分析 #

1. 当前校验逻辑存在漏洞：Slack的重试请求并非都会携带x-slack-retry-num头。比如当Slack因超时未收到响应时，会重复发送原始请求（不带重试头），此时当前校验逻辑无法识别。
2. 缺少幂等性保障：即使拦截了重试请求，若Slack重复发送同一事件的初始请求，业务逻辑仍会重复执行，破坏流程。

解决方案 #

1. 完善Slack重试请求的校验逻辑 #

Slack官方重试机制中，重试请求会同时携带x-slack-retry-num和x-slack-retry-reason头，且x-slack-retry-num从1开始递增。修改校验逻辑，确保正确识别所有官方重试请求：

from fastapi import Request

async def check_slack_retry(request: Request):
    retry_num = request.headers.get("x-slack-retry-num")
    retry_reason = request.headers.get("x-slack-retry-reason")
    # 识别Slack官方重试请求，直接返回200响应
    if retry_num is not None and retry_reason is not None:
        return {"status": "ok"}
    return None

将此函数作为slack-api的全局依赖或接口前置校验，确保重试请求被直接拦截。

2. 实现业务操作的幂等性（核心方案） #

针对Slack重复发送的无重试头请求，必须通过唯一事件标识实现幂等：

• 提取唯一标识：Slack每个事件都包含唯一的event_id字段（可从请求体的event对象中获取），也可使用x-slack-request-timestamp与x-slack-signature的组合作为唯一键。
• 记录已处理的事件：使用Redis（推荐，性能高效）或数据库临时表存储已处理的事件标识，设置合理的过期时间（比如24小时，根据业务场景调整）。
• 校验与执行逻辑：

import redis
from fastapi import Request

# 初始化Redis连接
redis_client = redis.Redis(host="your-redis-host", port=6379, db=0)

async def process_slack_event(request: Request):
    event_data = await request.json()
    event_id = event_data.get("event", {}).get("event_id")
    if not event_id:
        return {"status": "error", "msg": "缺失event_id标识"}
    
    # 检查事件是否已处理
    if redis_client.get(f"slack_event:{event_id}"):
        return {"status": "ok"}
    
    # 执行业务逻辑（调用backend-api）
    await call_backend_api(event_data)
    
    # 标记事件为已处理，设置24小时过期
    redis_client.setex(f"slack_event:{event_id}", 86400, "processed")
    return {"status": "ok"}

3. 优化响应速度，避免触发Slack重试 #

Slack要求在3秒内收到响应，否则会触发重试。因此slack-api应先返回200响应，再异步处理业务逻辑：

from fastapi import BackgroundTasks, Request

async def slack_webhook(request: Request, background_tasks: BackgroundTasks):
    # 立即返回响应，避免Slack超时重试
    response = {"status": "ok"}
    
    # 将业务逻辑加入后台任务异步执行
    event_data = await request.json()
    background_tasks.add_task(process_slack_event_async, event_data)
    
    return response

async def process_slack_event_async(event_data):
    # 这里执行原有的业务逻辑（调用backend-api等）
    # 同时加入幂等校验逻辑（参考步骤2）
    pass

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