LINE/WhatsApp Business API Webhook请求接收失败及ClientDisconnect异常排查

问题背景 #

集成LINE Messaging API和WhatsApp Business API时，通过ngrok实现公网映射，LINE服务器推送webhook请求时偶尔出现无法接收请求对象、触发ClientDisconnect异常（提示Client disconnected before the server could read the request body）。经排查是LINE服务器因超时断开连接，但自身服务器处理速度正常。已尝试提前返回200响应、流式读取请求等方案，均未解决；目前依赖ngrok inspect功能重试仅适用于测试环境，正式环境将切换自有域名，需彻底解决该问题。

解决方案 #

1. 优先返回响应，延后读取请求体 #

LINE的webhook超时阈值较短（默认5秒），当前代码在读取请求体时才触发异常，建议先直接返回200响应，再在后台异步读取请求内容，避免LINE服务器因等待响应超时断开。

修改后的核心代码：

@line_router.post("/callback")
async def handle_webhook(request: Request, background_tasks: BackgroundTasks):
    st = time.time()
    logger.info(f"Received LINE webhook request at {st}")

    # 立即返回200响应，阻断LINE服务器的超时逻辑
    response = JSONResponse(content={"status": "success", "message": "Webhook received"}, status_code=200)
    
    try:
        # 返回响应后再读取请求体
        body_bytes = await request.body()
        signature = request.headers.get("X-Line-Signature")
        
        et = time.time()
        logger.info(f"[Time] Processing Time: {et-st}")
        background_tasks.add_task(process_line_webhook, body_bytes, signature)
    except ClientDisconnect:
        logger.warning("Client disconnected before the request body could be fully read.")
        et = time.time()
        logger.info(f"[Time] Processing Time: {et-st}")
        # 正式环境需替换为持久化重试逻辑，比如Redis队列+定时任务
        return response

    return response

2. 利用LINE官方重试机制替代ngrok方案 #

LINE官方规定：若webhook响应超时或返回非2xx状态码，服务器会自动重试（最多3次，间隔1秒、10秒、1分钟）。结合这一特性：

• 无论是否读取到请求体，都优先返回200响应，避免LINE重复推送。
• 收到请求时记录X-Line-Request-Id（LINE为每个webhook生成的唯一标识），若读取失败，将该ID存入Redis待重试队列。
• 通过定时任务轮询队列，若LINE未自动重试，可触发本地重试逻辑。

3. 正式环境网络链路优化 #

切换自有域名后，从网络层面降低超时概率：

• 将服务器部署在靠近LINE亚太节点的区域（如新加坡），减少跨区域网络延迟。
• 调整Nginx等反向代理的超时设置，确保大于LINE的5秒超时阈值。
• 配置CDN或负载均衡，优化请求链路的稳定性。

4. 修复后台任务逻辑问题 #

当前process_failed_line_webhook存在两处逻辑错误：

• 处理事件循环时，第一个事件处理完就直接return，导致后续事件被忽略。
• 后台任务返回JSONResponse无意义，应移除返回值。

修复后的代码：

async def process_failed_line_webhook(body: bytes, signature: str):
    try:
        body_str = body.decode("utf-8")
        logger.info(f"[Decode] Incoming Webhook Data: {body_str}")
        logger.info(f"[Signature] Received LINE Callback request with signature: {signature}")

        if not line_client.verify_signature(body_str, signature if isinstance(signature, str) else signature[0]):
            logger.error("Invalid signature")
            return

        logger.info("Signature verified successfully.")
        data = json.loads(body_str)
        events = data.get("events", [])
        
        for event in events:
            logger.info(f"Incoming Event: {event}")
            if event.get("type") == "message":
                user_id = event.get("source", {}).get("userId")
                reply_token = event.get("replyToken")
                allowed_list = []  # 建议从配置/数据库读取
                if user_id not in allowed_list:
                    await line_client.send_text(reply_token, "Test")  # 确保send_text为异步调用
                else:
                    redis_data = {"message": json.dumps(event)}
                    redis_client.xadd(stream_name, redis_data)
                    logger.info(f"Message_Added_to_Redis : {redis_data}")
        logger.info("Messages processed successfully")

    except Exception:
        logger.exception("Error handling Line webhook: ")

内容的提问来源于stack exchange，提问作者Muhammad Faiz Raza