From 48c6f2a28f16f1e7da93b3640e87d26f3a83c64d Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Mon, 17 Aug 2026 01:05:37 +0800 Subject: [PATCH] =?UTF-8?q?feat(gateway):=20=E4=B8=8B=E5=8D=95=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E6=94=AF=E6=8C=81=20callback=5Furl=20=E7=BB=88?= =?UTF-8?q?=E7=BB=93=E7=B1=BB=E4=BA=8B=E4=BB=B6=E5=BC=82=E6=AD=A5=E9=80=9A?= =?UTF-8?q?=E7=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 通过 --- app/gateway/api/routes/orders.py | 9 +- app/gateway/callback.py | 119 +++++++++++ app/gateway/container.py | 2 + app/gateway/db.py | 17 +- app/gateway/main.py | 13 +- app/gateway/models.py | 24 ++- app/gateway/task_queue.py | 84 ++++++-- app/shared/config.py | 3 + docs/order-gateway.md | 50 ++++- tests/test_gateway_callback.py | 356 +++++++++++++++++++++++++++++++ 10 files changed, 649 insertions(+), 28 deletions(-) create mode 100644 app/gateway/callback.py create mode 100644 tests/test_gateway_callback.py diff --git a/app/gateway/api/routes/orders.py b/app/gateway/api/routes/orders.py index a3bf60d..87e3439 100644 --- a/app/gateway/api/routes/orders.py +++ b/app/gateway/api/routes/orders.py @@ -53,12 +53,19 @@ async def submit_order( 幂等：同一个 task_id 重复提交不新建任务，返回既有任务且 created=false—— 上游重发不会变成两单。下单不可逆，这是防重复下单的第一道闸。 + 可带 callback_url：任务到达终态（succeeded / failed / needs_human）或被置 + stale 时，网关向该地址 POST 一条 JSON 通知（best-effort 单次投递，§4.8）。 + 幂等重发不会更新既有任务的回调地址。 + 提交后任务为 queued。进度跟踪用 GET /api/orders/{task_id}（任务状态 + 完整状态历史）；任务状态词汇表：queued / leased / running / succeeded / failed / needs_human / stale。 """ data = await container.task_queue.submit( - task_id=payload.task_id, site=payload.site, intent=payload.intent + task_id=payload.task_id, + site=payload.site, + intent=payload.intent, + callback_url=payload.callback_url, ) return ApiResponse[SubmitOrderData]( success=True, msg="success", data=data, code=0 diff --git a/app/gateway/callback.py b/app/gateway/callback.py new file mode 100644 index 0000000..f1c713a --- /dev/null +++ b/app/gateway/callback.py @@ -0,0 +1,119 @@ +"""终结类事件回调：把任务结局 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() diff --git a/app/gateway/container.py b/app/gateway/container.py index e387705..a449cd8 100644 --- a/app/gateway/container.py +++ b/app/gateway/container.py @@ -38,5 +38,7 @@ class GatewayContainer: task_queue: TaskQueue query_queue: QueryQueue collector: object | None = field(default=None) # app.gateway.collector.OrderDiscoveryCollector + # 终结类事件回调（§4.8）：有 callback_url 的任务到达终态/变 stale 时 POST 通知上游 + notifier: object | None = field(default=None) # app.gateway.callback.CallbackNotifier sweep_task: asyncio.Task | None = None collector_task: asyncio.Task | None = None diff --git a/app/gateway/db.py b/app/gateway/db.py index d35ea1c..46aae09 100644 --- a/app/gateway/db.py +++ b/app/gateway/db.py @@ -25,6 +25,7 @@ CREATE TABLE IF NOT EXISTS tasks ( task_id TEXT PRIMARY KEY, site TEXT NOT NULL, intent_json TEXT NOT NULL, + callback_url TEXT, -- 终结类事件通知地址，见 docs/order-gateway.md §4.8 status TEXT NOT NULL, lease_owner TEXT, lease_expires_at TEXT, @@ -98,6 +99,7 @@ class TaskRow: task_id: str site: str intent_json: str + callback_url: str | None status: str lease_owner: str | None lease_expires_at: str | None @@ -180,6 +182,7 @@ def _row_to_task(row: aiosqlite.Row) -> TaskRow: task_id=row["task_id"], site=row["site"], intent_json=row["intent_json"], + callback_url=row["callback_url"], status=row["status"], lease_owner=row["lease_owner"], lease_expires_at=row["lease_expires_at"], @@ -256,9 +259,18 @@ class GatewayDB: self._conn = await aiosqlite.connect(str(self._db_path)) self._conn.row_factory = aiosqlite.Row await self._conn.executescript(SCHEMA) + await self._migrate() await self._conn.commit() logger.info("网关 DB 已就绪：%s", self._db_path) + async def _migrate(self) -> None: + """给既有库补新列（CREATE TABLE IF NOT EXISTS 不会改老表）""" + async with self.conn.execute("PRAGMA table_info(tasks)") as cur: + columns = {row["name"] for row in await cur.fetchall()} + if "callback_url" not in columns: + await self.conn.execute("ALTER TABLE tasks ADD COLUMN callback_url TEXT") + logger.info("迁移：tasks 表补充 callback_url 列") + async def close(self) -> None: if self._conn is not None: await self._conn.close() @@ -281,13 +293,14 @@ class GatewayDB: """插入新任务。返回 True=新建，False=task_id 已存在（幂等命中）""" try: await self.conn.execute( - "INSERT INTO tasks (task_id, site, intent_json, status, " + "INSERT INTO tasks (task_id, site, intent_json, callback_url, status, " "lease_owner, lease_expires_at, lease_count, created_at, updated_at) " - "VALUES (?, ?, ?, ?, NULL, NULL, 0, ?, ?)", + "VALUES (?, ?, ?, ?, ?, NULL, NULL, 0, ?, ?)", ( row.task_id, row.site, row.intent_json, + row.callback_url, row.status, row.created_at, row.updated_at, diff --git a/app/gateway/main.py b/app/gateway/main.py index 0273429..72f13d9 100644 --- a/app/gateway/main.py +++ b/app/gateway/main.py @@ -18,6 +18,7 @@ from app.gateway.api.routes.account import router as account_router from app.gateway.api.routes.health import router as health_router from app.gateway.api.routes.orders import router as orders_router from app.gateway.api.routes.queries import router as queries_router +from app.gateway.callback import CallbackNotifier from app.gateway.collector import OrderDiscoveryCollector from app.gateway.container import GatewayContainer from app.gateway.db import GatewayDB @@ -33,13 +34,18 @@ SWEEP_INTERVAL_SECONDS = 60 def build_container() -> GatewayContainer: - """构建网关容器：DB + 下单任务队列 + 账号只读查询队列""" + """构建网关容器：DB + 下单任务队列 + 账号只读查询队列 + 回调通知器""" settings = get_settings() db = GatewayDB(settings.gateway_db_path_resolved) + # 终结类事件回调（§4.8）：best-effort 单次投递，失败只记日志 + notifier = CallbackNotifier( + settings=settings, timeout_seconds=settings.callback_timeout_seconds + ) task_queue = TaskQueue( db, lease_ttl_seconds=settings.lease_ttl_seconds, worker_offline_alert_seconds=settings.worker_offline_alert_seconds, + notifier=notifier, ) query_queue = QueryQueue( db, @@ -53,7 +59,7 @@ def build_container() -> GatewayContainer: ) return GatewayContainer( settings=settings, db=db, task_queue=task_queue, - query_queue=query_queue, collector=collector, + query_queue=query_queue, collector=collector, notifier=notifier, ) @@ -131,6 +137,9 @@ async def lifespan(app: FastAPI): except asyncio.CancelledError: pass container.sweep_task = None + if container.notifier is not None: + # 等在途回调发完（各次发送有超时兜底），避免关停时静默丢通知 + await container.notifier.aclose() await container.db.close() diff --git a/app/gateway/models.py b/app/gateway/models.py index af2f804..31e29ac 100644 --- a/app/gateway/models.py +++ b/app/gateway/models.py @@ -7,8 +7,9 @@ intent 字段刻意保留成 `dict[str, Any]`——网关不解释下单意图 from __future__ import annotations from typing import Any +from urllib.parse import urlsplit -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, field_validator from app.shared.task_state import AccountQueryKind, OrderState, QueryStatus, TaskStatus @@ -49,6 +50,24 @@ class SubmitOrderRequest(BaseModel): "max_total_yen": 30000, }], ) + callback_url: str | None = Field( + default=None, + description="终结类事件的异步通知地址（http/https）。任务到达终态（succeeded / " + "failed / needs_human）或被置 stale 时，网关向该地址 POST 一条 JSON 通知，" + "上游可免去轮询。注意：幂等重发（created=false）不会更新既有任务的回调地址", + examples=["https://upstream.example.com/hooks/rakuten-order"], + ) + + @field_validator("callback_url") + @classmethod + def _validate_callback_url(cls, value: str | None) -> str | None: + """只接受 http/https 且带 host 的绝对 URL，其余当场 422""" + if value is None: + return None + parts = urlsplit(value) + if parts.scheme not in ("http", "https") or not parts.netloc: + raise ValueError("callback_url 必须是 http/https 绝对 URL") + return value class SubmitOrderData(BaseModel): @@ -194,6 +213,9 @@ class TaskDetail(BaseModel): task_id: str = Field(description="任务 ID（幂等键）") site: str = Field(description="站点标识（rakuten）") intent: dict[str, Any] = Field(description="下单意图原文") + callback_url: str | None = Field( + default=None, description="提交时登记的终结类事件通知地址；未登记为 null" + ) status: TaskStatus = Field( description="任务状态：queued / leased / running / succeeded / failed / needs_human / stale" ) diff --git a/app/gateway/task_queue.py b/app/gateway/task_queue.py index ab4cc84..f516989 100644 --- a/app/gateway/task_queue.py +++ b/app/gateway/task_queue.py @@ -25,6 +25,7 @@ from app.shared.task_state import ( TERMINAL_STATUSES, TaskStatus, ) +from app.gateway.callback import CallbackNotifier, stale_payload, terminal_payload from app.gateway.db import GatewayDB, ReportRow, TaskRow from app.gateway.models import ( GatewayHealthData, @@ -74,6 +75,7 @@ def _task_to_detail(task: TaskRow, latest_state: str | None, reports: list[Repor task_id=task.task_id, site=task.site, intent=task.intent, + callback_url=task.callback_url, status=TaskStatus(task.status), lease_owner=task.lease_owner, lease_expires_at=task.lease_expires_at, @@ -92,17 +94,37 @@ class TaskQueue: 保护下；长轮询等待通过 `self._cond` 唤醒。 """ - def __init__(self, db: GatewayDB, *, lease_ttl_seconds: int, worker_offline_alert_seconds: int): + def __init__( + self, + db: GatewayDB, + *, + lease_ttl_seconds: int, + worker_offline_alert_seconds: int, + notifier: CallbackNotifier | None = None, + ): self._db = db self._lease_ttl = lease_ttl_seconds self._worker_offline_alert_seconds = worker_offline_alert_seconds + # 终结类事件通知（§4.8）。None 表示完全关闭回调；测试可替换为记录型假实现 + self.notifier = notifier self._lock = asyncio.Lock() self._cond = asyncio.Condition(self._lock) # ---- 提交 ---- - async def submit(self, *, task_id: str | None, site: str, intent: dict) -> SubmitOrderData: - """上游提交下单意图。task_id 缺省时服务端生成；重复提交幂等""" + async def submit( + self, + *, + task_id: str | None, + site: str, + intent: dict, + callback_url: str | None = None, + ) -> SubmitOrderData: + """上游提交下单意图。task_id 缺省时服务端生成；重复提交幂等 + + callback_url 只在首次创建时写入；幂等重发（created=false）返回既有任务， + 不更新任何字段。 + """ import json tid = task_id or _generate_task_id() @@ -111,6 +133,7 @@ class TaskQueue: task_id=tid, site=site, intent_json=json.dumps(intent, ensure_ascii=False), + callback_url=callback_url, status=TaskStatus.QUEUED.value, lease_owner=None, lease_expires_at=None, @@ -152,7 +175,7 @@ class TaskQueue: async with self._lock: # 先把过期租约扫到 stale，避免「卡死的任务」堵住新任务 - await self._sweep_locked() + self._notify_stale(await self._sweep_locked()) await self._db.upsert_worker(worker_id, to_iso(utcnow())) @@ -272,18 +295,17 @@ class TaskQueue: ) now = to_iso(utcnow()) - inserted = await self._db.upsert_report( - ReportRow( - task_id=task_id, - state=state, - payable_yen=payable_yen, - pay_deadline=pay_deadline, - site_order_id=site_order_id, - evidence_ref=evidence_ref, - detail=detail, - reported_at=now, - ) + report_row = ReportRow( + task_id=task_id, + state=state, + payable_yen=payable_yen, + pay_deadline=pay_deadline, + site_order_id=site_order_id, + evidence_ref=evidence_ref, + detail=detail, + reported_at=now, ) + inserted = await self._db.upsert_report(report_row) final_status = status.value if terminal: @@ -298,6 +320,10 @@ class TaskQueue: ) # 任务终结可能让并发度 1 的闸门放开，唤醒等着的 lease self._cond.notify_all() + # 终结类事件回调（§4.8）：仅「本次真正把任务推入终态」才通知—— + # 已终结任务的后续 report（如付款后监控的 shipped/delivered）不重复通知 + if status not in TERMINAL_STATUSES: + self._notify_terminal(task, report_row, final_status) return ReportData(task_id=task_id, status=TaskStatus(final_status), recorded=inserted) @@ -355,13 +381,15 @@ class TaskQueue: async def sweep(self) -> int: """把过期的 leased/running 任务置 stale。返回清扫条数""" async with self._lock: - return await self._sweep_locked() + swept = await self._sweep_locked() + self._notify_stale(swept) + return len(swept) - async def _sweep_locked(self) -> int: - """锁内执行 sweep。调用方必须持有 self._lock""" + async def _sweep_locked(self) -> list[TaskRow]: + """锁内执行 sweep。调用方必须持有 self._lock。返回本次被置 stale 的任务""" active = await self._db.list_tasks_in_statuses(tuple(s.value for s in ACTIVE_STATUSES)) now = utcnow() - swept = 0 + swept: list[TaskRow] = [] for task in active: if task.lease_expires_at and parse_iso(task.lease_expires_at) < now: await self._db.update_task( @@ -370,7 +398,7 @@ class TaskQueue: clear_lease=True, updated_at=to_iso(now), ) - swept += 1 + swept.append(task) logger.warning( "租约过期，任务置 stale（不自动重投）：task_id=%s site=%s", task.task_id, @@ -382,6 +410,22 @@ class TaskQueue: self._cond.notify_all() return swept + # ---- 终结类事件回调（§4.8）---- + # notify 只是调度后台发送任务，持锁调用是安全的：真正的 HTTP 发送不碰这把锁 + + def _notify_terminal(self, task: TaskRow, report_row: ReportRow, final_status: str) -> None: + if self.notifier is not None and task.callback_url: + self.notifier.notify( + task.callback_url, terminal_payload(task, report_row, final_status) + ) + + def _notify_stale(self, swept: list[TaskRow]) -> None: + if self.notifier is None: + return + for task in swept: + if task.callback_url: + self.notifier.notify(task.callback_url, stale_payload(task)) + # ---- 查询 ---- async def get_task_detail(self, task_id: str) -> TaskDetail: diff --git a/app/shared/config.py b/app/shared/config.py index 1bb33df..268ad78 100644 --- a/app/shared/config.py +++ b/app/shared/config.py @@ -150,6 +150,9 @@ class Settings(BaseSettings): # worker 心跳超时阈值（秒）。网关 /health 据此判断 worker 是否失联： # 正常 worker 每 30 秒来一次 lease，超过该阈值未来 lease 即视为异常。 worker_offline_alert_seconds: int = 300 + # 终结类事件回调（callback_url）的单次 HTTP 超时（秒）。投递是 best-effort： + # 超时/失败只记日志不重试，权威状态以 GET /api/orders/{task_id} 为准。 + callback_timeout_seconds: float = 10.0 # ---- 账号只读查询通道（网关 + worker 两侧共用，见 docs/order-gateway.md §11）---- # 查询单租约 TTL（秒）。worker 领走后必须在此时间内回结果，否则网关把它 diff --git a/docs/order-gateway.md b/docs/order-gateway.md index 108b0a8..2f9914d 100644 --- a/docs/order-gateway.md +++ b/docs/order-gateway.md @@ -74,6 +74,7 @@ CREATE TABLE tasks ( task_id TEXT PRIMARY KEY, -- 幂等键，见 §4.2 site TEXT NOT NULL, -- rakuten（交易服务仅覆盖乐天市场） intent_json TEXT NOT NULL, -- 下单意图原文，gateway 不解释内容 + callback_url TEXT, -- 终结类事件通知地址（可选），见 §4.8 status TEXT NOT NULL, -- 见 §3 lease_owner TEXT, -- worker_id lease_expires_at TEXT, -- ISO8601 UTC @@ -116,14 +117,16 @@ CREATE TABLE workers ( "variant_id": "...", // 多规格商品必填 "options": {}, "max_total_yen": 30000 // 可选，覆盖本次的金额上限 - } + }, + "callback_url": "https://upstream.example.com/hooks/rakuten-order" // 可选，终结类事件通知，见 §4.8 } ``` 响应 `data`：`{"task_id": "...", "status": "queued", "created": true}` **幂等**：同一个 `task_id` 重复提交不新建任务，返回既有任务且 `created=false`。 -上游重发不会变成两单。 +上游重发不会变成两单。注意 `callback_url` 只在首次创建时写入，幂等重发不更新 +既有任务的任何字段——要换通知地址必须用新 `task_id`。 ### 4.3 GET /api/orders/lease — 本地长轮询领取 @@ -198,6 +201,48 @@ CREATE TABLE workers ( 付款期限监控**不在 gateway**：本地机 7×24 在线，那套逻辑放本地（§6）， gateway 不重复实现一遍定时器。 +### 4.8 回调通知（callback_url，2026-08-16） + +上游提交任务时可带 `callback_url`（http/https 绝对 URL，其余提交时 422）。 +登记后网关在**终结类事件**发生时向该地址 POST 一条 JSON 通知，上游免去轮询。 + +触发时机只有两类，每类一次： + +1. **任务到达终态**（`succeeded` / `failed` / `needs_human`）：worker 的 + `terminal=true` report 真正把任务推入终态的那一次。中间态 report + （in_cart / ordered / awaiting_payment 非 terminal）不通知；任务已终结后的 + 后续 report（如付款后监控的 shipped/delivered）也不重复通知。 +2. **任务被置 stale**：租约过期被 sweep 标记（见 §5）。上游收到后应走人工 + 排查 + reclaim 流程。 + +终态事件的 payload（无信封，直接是 JSON 对象）： + +```jsonc +{ + "task_id": "po-20260727-0001", + "site": "rakuten", + "event": "terminal", + "status": "succeeded", // 终态：succeeded / failed / needs_human + "state": "paid", // 本次 report 的订单状态 + "payable_yen": 12800, + "pay_deadline": "2026-07-30T14:59:00Z", + "site_order_id": "266123-20260727-0001234", + "evidence_ref": "po-20260727-0001/05-payment", + "detail": "付款完成", + "reported_at": "2026-07-27T09:15:00Z" +} +``` + +stale 事件的 payload：`{"task_id", "site", "event": "stale", "status": "stale", +"detail"}`。 + +投递语义是 **best-effort 单次尝试**：超时（`RAKUTEN_CALLBACK_TIMEOUT_SECONDS`， +默认 10 秒）、非 2xx、连接错误都只记网关日志，不重试、不阻塞 report/sweep +主流程。回调只是「省轮询」的提示，可能丢失——**权威状态仍以 +`GET /api/orders/{task_id}` 为准**，上游应保留对账轮询（可降低频率）。 +回调出站沿用统一代理策略（`app/shared/proxy.py`，按目标 host 判定 bypass）； +上游如需鉴权，可把 token 编进 callback_url 的 query 里自行校验。 + ## 5. 租约过期：绝不自动重投（本文最关键的一条） 常规任务队列在租约超时后会把任务放回队列重新分发。**这里必须禁止**：本地可能已经 @@ -280,6 +325,7 @@ gateway 侧： - `RAKUTEN_LEASE_TTL_SECONDS`（默认 300） - `RAKUTEN_LEASE_MAX_WAIT_SECONDS`（默认 60） - `RAKUTEN_WORKER_OFFLINE_ALERT_SECONDS`（默认 300） +- `RAKUTEN_CALLBACK_TIMEOUT_SECONDS`（默认 10，回调通知单次 HTTP 超时，见 §4.8） trading 侧新增： diff --git a/tests/test_gateway_callback.py b/tests/test_gateway_callback.py new file mode 100644 index 0000000..70bc88f --- /dev/null +++ b/tests/test_gateway_callback.py @@ -0,0 +1,356 @@ +"""终结类事件回调（callback_url）测试，对应 docs/order-gateway.md §4.8 + +三层覆盖： + +- API 层：提交校验（非法地址 422）、详情透出 callback_url、幂等重发不改地址、 + HTTP 全流程下 terminal report 恰好通知一次（注入记录型假 notifier）。 +- 队列层：中间态不通知、终结通知一次、终结后的后续 report 不重复通知、 + 租约过期 sweep 触发 stale 通知、未登记地址不通知。 +- 通知器层：真实发送路径（httpx.MockTransport）——成功送达、上游 5xx、 + 连接异常都不抛出、只记日志（best-effort）。 +""" +from __future__ import annotations + +import json +from pathlib import Path + +import aiosqlite +import httpx +import pytest +from fastapi.testclient import TestClient + +from app.gateway.callback import CallbackNotifier +from app.gateway.db import GatewayDB +from app.gateway.task_queue import TaskQueue +from app.shared.config import Settings, get_settings +from app.shared.task_state import OrderState, TaskStatus + +TOKEN = get_settings().bearer_token +AUTH = {"Authorization": f"Bearer {TOKEN}"} +CALLBACK_URL = "https://upstream.example.com/hooks/rakuten-order" + + +class FakeNotifier: + """记录型假通知器：与 CallbackNotifier 同接口，同步记录便于断言""" + + def __init__(self) -> None: + self.sent: list[tuple[str, dict]] = [] + + def notify(self, callback_url: str, payload: dict) -> None: + self.sent.append((callback_url, payload)) + + +# ---- API 层 ---- + + +@pytest.fixture +def gateway_client(tmp_path: Path, monkeypatch): + db_path = tmp_path / "gw.db" + monkeypatch.setenv("RAKUTEN_GATEWAY_DB_PATH", str(db_path)) + monkeypatch.setenv("RAKUTEN_ACCOUNT_DISCOVERY_ENABLED", "false") + get_settings.cache_clear() + try: + from app.gateway.main import create_app + + app = create_app() + with TestClient(app) as client: + yield client + finally: + get_settings.cache_clear() + + +@pytest.fixture +def fake_notifier(gateway_client): + """把容器里 TaskQueue 的通知器换成记录型假实现，用完还原""" + container = gateway_client.app.state.container + original = container.task_queue.notifier + fake = FakeNotifier() + container.task_queue.notifier = fake + yield fake + container.task_queue.notifier = original + + +def _submit(client, *, task_id: str, callback_url: str | None = CALLBACK_URL): + payload: dict = {"task_id": task_id, "site": "rakuten", "intent": {}} + if callback_url is not None: + payload["callback_url"] = callback_url + return client.post("/api/orders", json=payload, headers=AUTH) + + +def test_submit_with_callback_url_roundtrip(gateway_client): + response = _submit(gateway_client, task_id="t1") + assert response.status_code == 200 + assert response.json()["data"]["created"] is True + + detail = gateway_client.get("/api/orders/t1", headers=AUTH).json()["data"] + assert detail["callback_url"] == CALLBACK_URL + + +def test_submit_without_callback_url_defaults_to_null(gateway_client): + _submit(gateway_client, task_id="t1", callback_url=None) + detail = gateway_client.get("/api/orders/t1", headers=AUTH).json()["data"] + assert detail["callback_url"] is None + + +@pytest.mark.parametrize( + "bad_url", + [ + "ftp://upstream.example.com/hook", # 非 http/https scheme + "not-a-url", # 没有 scheme 与 host + "https://", # 有 scheme 没 host + ], +) +def test_submit_rejects_invalid_callback_url(gateway_client, bad_url): + response = _submit(gateway_client, task_id="t1", callback_url=bad_url) + assert response.status_code == 422 + + +def test_idempotent_resubmit_keeps_original_callback_url(gateway_client): + """同 task_id 重发带不同 callback_url：不新建、不更新地址""" + _submit(gateway_client, task_id="t1", callback_url=CALLBACK_URL) + r2 = _submit(gateway_client, task_id="t1", callback_url="https://other.example.com/hook") + assert r2.json()["data"]["created"] is False + + detail = gateway_client.get("/api/orders/t1", headers=AUTH).json()["data"] + assert detail["callback_url"] == CALLBACK_URL + + +def test_terminal_report_sends_exactly_one_callback(gateway_client, fake_notifier): + _submit(gateway_client, task_id="t1") + gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH) + + response = gateway_client.post( + "/api/orders/t1/report", + json={ + "worker_id": "w1", + "state": OrderState.PAID.value, + "payable_yen": 9800, + "site_order_id": "ord-1", + "terminal": True, + "terminal_status": TaskStatus.SUCCEEDED.value, + "detail": "付款完成", + }, + headers=AUTH, + ) + assert response.status_code == 200 + + assert len(fake_notifier.sent) == 1 + url, payload = fake_notifier.sent[0] + assert url == CALLBACK_URL + assert payload["event"] == "terminal" + assert payload["task_id"] == "t1" + assert payload["site"] == "rakuten" + assert payload["status"] == TaskStatus.SUCCEEDED.value + assert payload["state"] == OrderState.PAID.value + assert payload["payable_yen"] == 9800 + assert payload["site_order_id"] == "ord-1" + assert payload["detail"] == "付款完成" + assert payload["reported_at"] + + +def test_non_terminal_report_sends_no_callback(gateway_client, fake_notifier): + _submit(gateway_client, task_id="t1") + gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH) + gateway_client.post( + "/api/orders/t1/report", + json={"worker_id": "w1", "state": OrderState.IN_CART.value}, + headers=AUTH, + ) + assert fake_notifier.sent == [] + + +def test_report_after_terminal_sends_no_more_callbacks(gateway_client, fake_notifier): + """任务终结后的后续 report（付款后监控）与幂等重报都不再通知""" + _submit(gateway_client, task_id="t1") + gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH) + terminal_report = { + "worker_id": "w1", + "state": OrderState.PAID.value, + "terminal": True, + "terminal_status": TaskStatus.SUCCEEDED.value, + } + gateway_client.post("/api/orders/t1/report", json=terminal_report, headers=AUTH) + assert len(fake_notifier.sent) == 1 + + # 同一 (task_id, state) 幂等重报 + 终结后的监控上报，都不应再触发 + gateway_client.post("/api/orders/t1/report", json=terminal_report, headers=AUTH) + gateway_client.post( + "/api/orders/t1/report", + json={"worker_id": "w1", "state": OrderState.SHIPPED.value}, + headers=AUTH, + ) + assert len(fake_notifier.sent) == 1 + + +def test_needs_human_inferred_terminal_sends_callback(gateway_client, fake_notifier): + """terminal=true 不显式给终态时按 state 推断（非 paid/cancelled → needs_human）""" + _submit(gateway_client, task_id="t1") + gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH) + gateway_client.post( + "/api/orders/t1/report", + json={ + "worker_id": "w1", + "state": OrderState.AWAITING_PAYMENT.value, + "terminal": True, + "detail": "弹了 3DS", + }, + headers=AUTH, + ) + assert len(fake_notifier.sent) == 1 + _, payload = fake_notifier.sent[0] + assert payload["status"] == TaskStatus.NEEDS_HUMAN.value + + +# ---- 队列层 ---- + + +@pytest.fixture +async def queue(tmp_path: Path): + db = GatewayDB(tmp_path / "gw.db") + await db.start() + fake = FakeNotifier() + q = TaskQueue( + db, lease_ttl_seconds=0, worker_offline_alert_seconds=300, notifier=fake + ) + yield q, fake + await db.close() + + +async def test_sweep_stale_sends_callback(queue): + """租约过期被 sweep 置 stale 时触发 stale 通知（上游需要人工 reclaim）""" + q, fake = queue + await q.submit( + task_id="t1", site="rakuten", intent={}, callback_url=CALLBACK_URL + ) + # lease_ttl=0：领走即过期，下一次 sweep 置 stale + await q.lease(worker_id="w1", wait=0, site=None, max_wait=60) + + swept = await q.sweep() + assert swept == 1 + assert len(fake.sent) == 1 + url, payload = fake.sent[0] + assert url == CALLBACK_URL + assert payload["event"] == "stale" + assert payload["status"] == TaskStatus.STALE.value + assert payload["task_id"] == "t1" + + +async def test_sweep_without_callback_url_sends_nothing(queue): + q, fake = queue + await q.submit(task_id="t1", site="rakuten", intent={}) + await q.lease(worker_id="w1", wait=0, site=None, max_wait=60) + + swept = await q.sweep() + assert swept == 1 + assert fake.sent == [] + + +async def test_terminal_without_callback_url_sends_nothing(queue): + q, fake = queue + await q.submit(task_id="t1", site="rakuten", intent={}) + await q.lease(worker_id="w1", wait=0, site=None, max_wait=60) + await q.report( + "t1", + worker_id="w1", + state=OrderState.PAID.value, + payable_yen=None, + pay_deadline=None, + site_order_id=None, + evidence_ref=None, + detail="", + terminal=True, + terminal_status=None, + ) + assert fake.sent == [] + + +# ---- DB 迁移 ---- + + +async def test_start_migrates_tasks_table_without_callback_url(tmp_path: Path): + """既有库（无 callback_url 列的老表）启动时补列，原数据不丢""" + db_path = tmp_path / "old.db" + conn = await aiosqlite.connect(str(db_path)) + await conn.execute( + "CREATE TABLE tasks (" + "task_id TEXT PRIMARY KEY, site TEXT NOT NULL, intent_json TEXT NOT NULL, " + "status TEXT NOT NULL, lease_owner TEXT, lease_expires_at TEXT, " + "lease_count INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, " + "updated_at TEXT NOT NULL)" + ) + await conn.execute( + "INSERT INTO tasks VALUES ('t1', 'rakuten', '{}', 'queued', " + "NULL, NULL, 0, '2026-08-16T00:00:00Z', '2026-08-16T00:00:00Z')" + ) + await conn.commit() + await conn.close() + + db = GatewayDB(db_path) + await db.start() + try: + task = await db.get_task("t1") + assert task is not None + assert task.callback_url is None + finally: + await db.close() + + +# ---- 通知器层（真实发送路径，MockTransport）---- + + +def _bare_settings() -> Settings: + """不读 .env 的最小配置，通知器只要代理策略字段""" + return Settings(_env_file=None) + + +async def test_notifier_delivers_payload(): + received: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + received.append(request) + return httpx.Response(200) + + notifier = CallbackNotifier( + settings=_bare_settings(), + timeout_seconds=5, + client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) + payload = {"task_id": "t1", "event": "terminal", "status": "succeeded"} + notifier.notify(CALLBACK_URL, payload) + await notifier.aclose() # 等在途通知发完 + + assert len(received) == 1 + assert str(received[0].url) == CALLBACK_URL + assert json.loads(received[0].content) == payload + + +async def test_notifier_swallows_upstream_5xx(caplog): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(500) + + notifier = CallbackNotifier( + settings=_bare_settings(), + timeout_seconds=5, + client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) + with caplog.at_level("WARNING", logger="app.gateway.callback"): + notifier.notify(CALLBACK_URL, {"task_id": "t1"}) + await notifier.aclose() # 不抛出 + + assert any("非成功状态" in rec.message for rec in caplog.records) + + +async def test_notifier_swallows_connection_error(caplog): + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("connection refused", request=request) + + notifier = CallbackNotifier( + settings=_bare_settings(), + timeout_seconds=5, + client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) + with caplog.at_level("WARNING", logger="app.gateway.callback"): + notifier.notify(CALLBACK_URL, {"task_id": "t1"}) + await notifier.aclose() # 不抛出 + + assert any("发送失败" in rec.message for rec in caplog.records)