用请求截止时间预算串联网关与下游超时

问题场景

网关收到请求后，如果只检查 X-Request-Deadline 是否存在，却不把剩余时间传给下游调用，就会出现一种常见错配：客户端已经接近超时，网关仍按固定超时等待慢依赖。结果是连接和协程被占用，客户端先断开，网关日志里只留下一个泛化错误。

更合理的做法是把截止时间当作预算：网关解析绝对毫秒时间戳，计算当前剩余毫秒；进入下游前扣除网关自身保留时间；再按比例把预算转换成 HTTP 客户端超时。剩余预算低于最小阈值时，不再发起下游调用，直接返回 504，并记录预算耗尽发生在哪个阶段、哪个依赖、预算和耗时是多少。

下面的示例用 FastAPI 实现网关中间件和下游调用封装，并用同进程的 /slow 端点模拟慢依赖。示例参数仅用于说明机制，不是通用最优值。

可运行的最小代码示例

保存为 main.py。

import asyncio
import json
import logging
import time

import httpx
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()
logger = logging.getLogger('gateway')
logging.basicConfig(level=logging.INFO)

GATEWAY_MIN_REMAINING_MS = 30
GATEWAY_RESERVE_MS = 10
DOWNSTREAM_RATIO = 0.6
MIN_DOWNSTREAM_MS = 20


def now_ms():
    return int(time.time() * 1000)


class BudgetExhausted(Exception):
    def __init__(self, detail):
        super().__init__(detail.get('reason', 'budget_exhausted'))
        self.detail = detail


@app.middleware('http')
async def deadline_budget_middleware(request: Request, call_next):
    if not request.url.path.startswith('/gateway'):
        return await call_next(request)
    raw = request.headers.get('X-Request-Deadline')
    if not raw:
        return JSONResponse(status_code=400, content={'error': 'missing_deadline'})
    try:
        deadline_ms = int(raw)
    except ValueError:
        return JSONResponse(status_code=400, content={'error': 'invalid_deadline'})
    remaining_ms = deadline_ms - now_ms()
    if remaining_ms <= GATEWAY_MIN_REMAINING_MS:
        detail = {
            'stage': 'gateway',
            'reason': 'remaining_below_gateway_min',
            'remaining_ms': remaining_ms,
            'deadline_ms': deadline_ms,
        }
        logger.warning('budget_exhausted %s', json.dumps(detail, ensure_ascii=False))
        return JSONResponse(status_code=504, content={'error': 'deadline_exhausted', **detail})
    request.state.deadline_ms = deadline_ms
    request.state.remaining_ms = remaining_ms
    return await call_next(request)


async def call_downstream(request, url, name, ratio):
    deadline_ms = request.state.deadline_ms
    remaining_ms = deadline_ms - now_ms() - GATEWAY_RESERVE_MS
    if remaining_ms <= MIN_DOWNSTREAM_MS:
        detail = {
            'stage': 'downstream',
            'dependency': name,
            'reason': 'remaining_below_downstream_min',
            'remaining_ms': remaining_ms,
            'deadline_ms': deadline_ms,
        }
        logger.warning('budget_exhausted %s', json.dumps(detail, ensure_ascii=False))
        raise BudgetExhausted(detail)
    budget_ms = max(MIN_DOWNSTREAM_MS, int(remaining_ms * ratio))
    timeout = httpx.Timeout(budget_ms / 1000.0)
    started_ms = now_ms()
    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            response = await client.get(
                url,
                headers={'X-Request-Deadline': str(deadline_ms)},
            )
        response.raise_for_status()
        return response.json()
    except httpx.TimeoutException as exc:
        elapsed_ms = now_ms() - started_ms
        detail = {
            'stage': 'downstream',
            'dependency': name,
            'reason': 'downstream_timeout',
            'budget_ms': budget_ms,
            'elapsed_ms': elapsed_ms,
            'deadline_ms': deadline_ms,
        }
        logger.warning('budget_exhausted %s', json.dumps(detail, ensure_ascii=False))
        raise BudgetExhausted(detail) from exc


@app.get('/gateway/order')
async def gateway_order(request: Request):
    try:
        primary = await call_downstream(
            request,
            'http://127.0.0.1:8000/slow?delay_ms=120',
            'primary',
            DOWNSTREAM_RATIO,
        )
        secondary = await call_downstream(
            request,
            'http://127.0.0.1:8000/slow?delay_ms=30',
            'secondary',
            0.3,
        )
        return {'primary': primary, 'secondary': secondary}
    except BudgetExhausted as exc:
        return JSONResponse(
            status_code=504,
            content={'error': 'deadline_exhausted', **exc.detail},
        )


@app.get('/slow')
async def slow(delay_ms: int = 0):
    await asyncio.sleep(max(0, delay_ms) / 1000.0)
    return {'ok': True, 'delay_ms': delay_ms}

运行方法

安装依赖并启动服务：

pip install fastapi uvicorn httpx
uvicorn main:app --port 8000

另开一个终端，用标准库脚本构造不同预算验证。offset_ms 分别模拟预算充足、网关剩余不足、下游超时三种情况：

python - <<'PY'
import time
import urllib.error
import urllib.request

for offset_ms in (1000, 20, 150):
    deadline = int(time.time() * 1000) + offset_ms
    req = urllib.request.Request(
        'http://127.0.0.1:8000/gateway/order',
        headers={'X-Request-Deadline': str(deadline)},
    )
    try:
        with urllib.request.urlopen(req, timeout=5) as resp:
            print(offset_ms, resp.status, resp.read().decode())
    except urllib.error.HTTPError as exc:
        print(offset_ms, exc.code, exc.read().decode())
PY

预期结果：预算约 1000ms 时能返回 200；预算约 20ms 时中间件直接返回 504，原因包含 remaining_below_gateway_min；预算约 150ms 时主依赖分配到的超时低于其 120ms 延迟，返回 504，原因包含 downstream_timeout。日志中应能看到 budget_exhausted 及对应 JSON 字段。

边界与限制

X-Request-Deadline 使用绝对毫秒时间戳，依赖客户端与服务端时钟同步。跨主机部署时应使用统一时间源；同时要限制可接受的最大截止时间，避免异常大值绕过预算约束。示例未加入最大上限。

比例分配只是演示。真实系统中，关键依赖和非关键依赖应有不同权重；并行调用时需要为每个依赖单独计算预算，不能把同一份剩余时间重复分配。顺序调用则会自然累计消耗预算，后调用的可用时间会变少。

HTTP 客户端超时能覆盖连接、读取等阶段，但 DNS 解析、连接池等待和事件循环调度仍可能让实际耗时略超预算。504 表示预算不足或下游超时，不等同于业务失败，调用方仍需结合幂等、重试和降级策略处理。示例把慢依赖和网关放在同一进程同一端口，仅用于验证预算逻辑；生产环境应独立部署，并持续采集日志和指标。