q792602257
48c6f2a28f
- POST /api/orders 新增可选 callback_url（仅 http/https，其余 422） - 仅终结类事件各通知一次：terminal report 推入终态（succeeded/failed/ needs_human）、租约过期被 sweep 置 stale；中间态与终结后的监控上报不通知 - 幂等重发不更新既有任务的回调地址；投递 best-effort 单次尝试，失败只记日志 - CallbackNotifier 发后不管（create_task + 在途任务强引用），关停等在途发完； 生产路径按回调地址逐次构造客户端，满足统一出站代理策略（test_proxy.py） - tasks 表加 callback_url 列，GatewayDB.start() 内置迁移兼容既有库 - 新增 RAKUTEN_CALLBACK_TIMEOUT_SECONDS（默认 10）；文档补 §4.8； openapi.json 重新导出（gitignore 未跟踪）；新增 17 条测试，全量 492 通过
120 lines
4.8 KiB
Python
120 lines
4.8 KiB
Python
"""终结类事件回调：把任务结局 POST 到上游登记的通知地址
|
|
|
|
触发时机只有两类（docs/order-gateway.md §4.8）：
|
|
|
|
- worker 的 terminal report 把任务推入终态（succeeded / failed / needs_human）
|
|
- 租约过期被 sweep 置 stale（任务卡住，需要人工 reclaim）
|
|
|
|
投递是 best-effort：单次尝试，失败只记日志、不重试。回调只是「省轮询」的提示，
|
|
权威状态仍以 GET /api/orders/{task_id} 为准。任何异常都不允许逃出发送路径，
|
|
不能影响 report / sweep 主流程。
|
|
|
|
出站遵循统一代理策略（app/shared/proxy.py）：回调地址逐任务不同，而代理 bypass
|
|
是按目标 host 判定的，所以客户端按发送逐次构造，不能在建通知器时定死。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
import httpx
|
|
|
|
from app.shared.proxy import httpx_client_options
|
|
|
|
if TYPE_CHECKING:
|
|
from app.gateway.db import ReportRow, TaskRow
|
|
from app.shared.config import Settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def terminal_payload(task: TaskRow, report: ReportRow, final_status: str) -> dict[str, Any]:
|
|
"""terminal report 触发的通知：任务终态 + 本次上报的字段（金额/单号/证据路径等）"""
|
|
return {
|
|
"task_id": task.task_id,
|
|
"site": task.site,
|
|
"event": "terminal",
|
|
"status": final_status,
|
|
"state": report.state,
|
|
"payable_yen": report.payable_yen,
|
|
"pay_deadline": report.pay_deadline,
|
|
"site_order_id": report.site_order_id,
|
|
"evidence_ref": report.evidence_ref,
|
|
"detail": report.detail,
|
|
"reported_at": report.reported_at,
|
|
}
|
|
|
|
|
|
def stale_payload(task: TaskRow) -> dict[str, Any]:
|
|
"""租约过期触发的通知：任务卡住，需人工 reclaim（绝不自动重投，见规格 §5）"""
|
|
return {
|
|
"task_id": task.task_id,
|
|
"site": task.site,
|
|
"event": "stale",
|
|
"status": "stale",
|
|
"detail": "租约过期，任务已置 stale，需人工 reclaim",
|
|
}
|
|
|
|
|
|
class CallbackNotifier:
|
|
"""把终结类事件 POST 到上游 callback_url（发后不管，失败只记日志）
|
|
|
|
`notify` 只做调度，真正的 HTTP 发送在后台任务里完成，调用方（report / sweep）
|
|
不被回调的耗时阻塞。在途任务用 `self._pending` 持有强引用，避免事件循环只持
|
|
弱引用导致任务被提前回收；`aclose` 时会等这些在途通知发完（各自有超时兜底）。
|
|
|
|
`client` 仅供测试注入（如 MockTransport）；生产路径为 None，按回调地址
|
|
逐次构造带代理策略的临时客户端（回调是低频事件，代价可忽略）。
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
settings: Settings,
|
|
timeout_seconds: float,
|
|
client: httpx.AsyncClient | None = None,
|
|
):
|
|
self._settings = settings
|
|
self._timeout = timeout_seconds
|
|
self._client = client
|
|
self._pending: set[asyncio.Task[None]] = set()
|
|
|
|
def notify(self, callback_url: str, payload: dict[str, Any]) -> None:
|
|
"""调度一次后台发送。调用方不等待，回调慢/挂不阻塞任务主流程"""
|
|
task = asyncio.create_task(self._send(callback_url, payload))
|
|
self._pending.add(task)
|
|
task.add_done_callback(self._pending.discard)
|
|
|
|
async def _send(self, callback_url: str, payload: dict[str, Any]) -> None:
|
|
task_id = payload.get("task_id")
|
|
try:
|
|
if self._client is not None:
|
|
response = await self._client.post(callback_url, json=payload)
|
|
else:
|
|
async with httpx.AsyncClient(
|
|
timeout=self._timeout,
|
|
**httpx_client_options(self._settings, target_url=callback_url),
|
|
) as client:
|
|
response = await client.post(callback_url, json=payload)
|
|
except Exception as exc: # noqa: BLE001
|
|
# 回调失败不允许影响任务主流程：记日志后丢弃，上游可靠轮询对账兜底
|
|
logger.warning(
|
|
"回调通知发送失败（已丢弃）：task_id=%s url=%s err=%s", task_id, callback_url, exc
|
|
)
|
|
return
|
|
if response.status_code >= 400:
|
|
logger.warning(
|
|
"回调通知返回非成功状态（已丢弃）：task_id=%s url=%s status=%s",
|
|
task_id, callback_url, response.status_code,
|
|
)
|
|
return
|
|
logger.info("回调通知已送达：task_id=%s url=%s", task_id, callback_url)
|
|
|
|
async def aclose(self) -> None:
|
|
"""关停：等在途通知发完（各次发送有超时兜底），再关 HTTP 客户端"""
|
|
if self._pending:
|
|
await asyncio.gather(*self._pending, return_exceptions=True)
|
|
if self._client is not None:
|
|
await self._client.aclose()
|