From 8955a0d1189eb55025d48c7bb148aafca9fc6125 Mon Sep 17 00:00:00 2001 From: zhx Date: Tue, 12 May 2026 20:07:49 +0800 Subject: [PATCH] =?UTF-8?q?=E6=98=8E=E7=A1=AEcomand=5Fid=E8=AF=AD=E4=B9=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 2 +- README.md | 4 +- docs/VIP系统架构.md | 2 +- docs/api-test-harness.html | 4 +- docs/openapi/activity.swagger.yaml | 1 + docs/openapi/gateway.swagger.yaml | 110 ++--- docs/openapi/room.swagger.yaml | 2 + docs/主播公会BD架构.md | 2 +- docs/币商库存授信架构.md | 3 +- docs/房间Outbox补偿开发.md | 393 +++++------------- docs/接口清单.md | 12 +- docs/服务基础技术设计.md | 2 +- docs/每日任务系统架构.md | 4 +- docs/游戏接入架构.md | 2 +- docs/登录注册技术设计.md | 2 +- docs/语音房功能清单.md | 34 +- docs/语音房基础闭环实现.md | 26 +- docs/语音房客户端接口流程.md | 2 +- docs/资源组与礼物架构.md | 2 +- docs/进房与重连开发需求.md | 7 +- server/admin/internal/middleware/request.go | 5 +- .../internal/transport/http/handler.go | 14 +- .../transport/http/httpkit/response.go | 9 +- .../http/manager_center_handler_test.go | 7 +- .../http/my_page_overview_handler_test.go | 2 +- .../internal/transport/http/response_test.go | 82 ++-- .../internal/transport/http/room_handler.go | 8 +- .../internal/transport/http/task_handler.go | 4 +- .../internal/room/service/create_room.go | 4 + .../internal/room/service/helpers.go | 2 +- .../internal/room/service/service_test.go | 1 + .../room-service/internal/router/directory.go | 12 +- .../internal/router/directory_test.go | 61 +++ .../internal/router/redis_directory.go | 18 +- 34 files changed, 342 insertions(+), 503 deletions(-) create mode 100644 services/room-service/internal/router/directory_test.go diff --git a/AGENTS.md b/AGENTS.md index 2084e4e8..bcd0e714 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -44,7 +44,7 @@ - 外部 HTTP JSON 入口在 `gateway-service`。 - `gateway-service` 的 `/api/v1` 业务响应必须使用 `{code,message,request_id,data}` envelope；healthcheck 不套业务响应格式。 -- `request_id` 是链路追踪字段，必须透传到内部 gRPC `RequestMeta.request_id`；不要把它当幂等键使用。 +- `request_id` 是后端生成的链路追踪字段，必须透传到内部 gRPC `RequestMeta.request_id`；不要把它当幂等键使用。 - 外部实时消息入口在腾讯云 IM SDK；本仓库只提供 `/api/v1/im/usersig` 签发登录票据。 - 内部服务通信统一 gRPC + protobuf，Go 代码通过 `hyapp.local/api/proto/...` 引用生成包。 - 当前处于开发阶段，不保留历史协议；Proto/API/DB 可以按当前事实直接调整。 diff --git a/README.md b/README.md index 89e47a12..2bdfda23 100644 --- a/README.md +++ b/README.md @@ -101,7 +101,7 @@ Authorization: Bearer - 客户端先经 gateway 调 `JoinRoom`，让用户进入 `room-service` presence。 - 客户端再用腾讯云 IM SDK 登录并加入 `room_id` 对应的群组。 - `gateway-service` 创建房间时从 access token 取当前 `user_id` 写入 `RequestMeta.actor_user_id`；`room-service` 由该 actor 收敛 owner 和默认 host，外部 CreateRoom 请求体不接收 `owner_user_id` 和 `host_user_id`。 -- 房间 HTTP 命令支持请求体 `command_id` 作为幂等键；`request_id` 只做链路追踪。 +- `X-Request-ID`/`request_id` 由后端生成，只做链路追踪；写操作的 `command_id` 由前端按用户动作生成，用于服务端幂等。 - `room-service` 在 `CreateRoom` 时通过腾讯云 IM REST API 创建同名群组。 - `room-service` 在上下麦、禁言、踢人、送礼等命令提交后，通过腾讯云 IM REST API 发送房间系统自定义消息。 - `gateway-service` 已提供 `/api/v1/tencent-im/callback`，支持腾讯云 IM 入群前和发言前回调，并分别回查 `VerifyRoomPresence` 和 `CheckSpeakPermission`。 @@ -180,7 +180,7 @@ services/cron-service/ background scheduling and owner-service batch trigger ## Current Limits - 腾讯云 IM 回调入口已经落在 gateway；线上仍必须在腾讯云 IM 控制台启用 `Group.CallbackBeforeApplyJoinGroup` 和 `Group.CallbackBeforeSendMsg`，并配置公网 callback URL 与鉴权 token。 -- 房间命令未传 `command_id` 时，gateway 会自动生成新值；客户端 HTTP 重试同一业务动作时必须复用原 `command_id` 才能命中 room-service command log 幂等。 +- 前端对同一次用户写动作的 HTTP 重试必须复用同一个 `command_id`；`X-Request-ID` 不需要也不应该由前端传入。 - `wallet-service` App 运行路径使用 MySQL 扣费事务；本地需要先通过 compose initdb 建好 `hyapp_wallet`。 - `activity-service` 已承接每日任务和消息/活动基础能力；`room_outbox` 到 activity/audit 的实际消费链路仍需按独立消费位点补齐。 - `room-service` App 已启动 `room_outbox` 补偿 worker；该状态只表示腾讯云 IM 系统消息补偿结果，activity/audit 后续必须使用独立消费位点。 diff --git a/docs/VIP系统架构.md b/docs/VIP系统架构.md index 41da143e..ae6f1d8e 100644 --- a/docs/VIP系统架构.md +++ b/docs/VIP系统架构.md @@ -25,7 +25,7 @@ VIP 属于 `wallet-service` 的账务和权益域，不放在 `user-service`。 服务职责： -- `gateway-service`：提供 App HTTP API，做鉴权、参数校验、request_id/command_id 透传，调用 `wallet-service`。 +- `gateway-service`：提供 App HTTP API，做鉴权、参数校验、生成 `request_id`、透传客户端 `command_id`，调用 `wallet-service`。 - `wallet-service`：拥有 VIP 等级配置、用户 VIP 状态、VIP 购买订单、金币扣费、资源组发放、VIP 事件 outbox。 - `user-service`：只在用户资料聚合时读取 VIP 摘要，不拥有 VIP 状态。 - `room-service`：需要展示 VIP 标识时读取用户资料/VIP 投影，不参与 VIP 购买主链路。 diff --git a/docs/api-test-harness.html b/docs/api-test-harness.html index 34cd9916..ac4c6773 100644 --- a/docs/api-test-harness.html +++ b/docs/api-test-harness.html @@ -2350,7 +2350,7 @@ function headersFor(source, withJson = true) { const c = cfg(); - const headers = { 'X-Request-ID': commandId('req') }; + const headers = {}; if (withJson) headers['Content-Type'] = 'application/json'; if (c.appCode) headers['X-App-Code'] = c.appCode; const token = source === 'admin' ? c.adminToken : c.accessToken; @@ -2459,7 +2459,7 @@ if (p.endsWith('/users/me/country/change')) return { country: c.country || 'US' }; if (p.endsWith('/users/me/display-id/change')) return { new_display_user_id: c.displayUserId }; if (p.endsWith('/users/me/display-id/pretty/apply')) return { pretty_display_user_id: c.displayUserId }; - if (p.endsWith('/rooms/create')) return { mode: 'voice', room_name: 'Smoke Room', room_avatar: '', room_description: 'local smoke room' }; + if (p.endsWith('/rooms/create')) return { command_id: commandId('cmd_create'), mode: 'voice', room_name: 'Smoke Room', room_avatar: '', room_description: 'local smoke room' }; if (p.endsWith('/rooms/join')) return { command_id: commandId('cmd_join'), room_id: c.roomId, role: 'audience' }; if (p.endsWith('/rooms/heartbeat')) return { command_id: commandId('cmd_heartbeat'), room_id: c.roomId }; if (p.endsWith('/rooms/leave')) return { command_id: commandId('cmd_leave'), room_id: c.roomId }; diff --git a/docs/openapi/activity.swagger.yaml b/docs/openapi/activity.swagger.yaml index 4594f574..660ccb50 100644 --- a/docs/openapi/activity.swagger.yaml +++ b/docs/openapi/activity.swagger.yaml @@ -536,6 +536,7 @@ definitions: $ref: "#/definitions/RequestMeta" command_id: type: string + description: 客户端或操作端前端按单次写动作生成的幂等命令 ID，同一动作重试必须复用。 message_type: type: string enum: diff --git a/docs/openapi/gateway.swagger.yaml b/docs/openapi/gateway.swagger.yaml index d05b2841..0ebc35c0 100644 --- a/docs/openapi/gateway.swagger.yaml +++ b/docs/openapi/gateway.swagger.yaml @@ -4,7 +4,7 @@ info: version: "1.0.0" description: | gateway-service 对外暴露 HTTP + JSON API。 - `/api/v1/**` 业务接口统一返回 `code/message/request_id/data` 外壳，并通过 `X-Request-ID` 贯穿 HTTP 响应和内部 gRPC RequestMeta。 + `/api/v1/**` 业务接口统一返回 `code/message/request_id/data` 外壳；`request_id` 和响应 `X-Request-ID` 均由服务端生成，并透传到内部 gRPC RequestMeta。 `/healthz/**` 是健康检查接口，不使用业务响应外壳。 host: localhost:13000 basePath: / @@ -52,12 +52,6 @@ securityDefinitions: in: header description: 使用 `Bearer `；JWT 使用 HS256，claims 中的 `user_id` 是数值型用户 ID 的字符串，避免 64 位 ID 精度丢失。 parameters: - RequestIDHeader: - name: X-Request-ID - in: header - required: false - type: string - description: 可选追踪 ID；未传时 gateway 自动生成，并返回在响应 header 和 JSON `request_id`。该字段不是幂等键；需要幂等的写命令使用各自命令 ID，其中创建房间由 gateway 生成内部 `command_id`。 DisplayUserIDPath: name: display_user_id in: path @@ -187,7 +181,6 @@ paths: description: 账号就是用户当前有效短号。 operationId: loginAccountPassword parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -215,7 +208,6 @@ paths: summary: 三方登录或注册 operationId: loginThirdParty parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -245,7 +237,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -275,7 +266,6 @@ paths: summary: 刷新 token operationId: refreshToken parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -303,7 +293,6 @@ paths: summary: 登出并失效 refresh session operationId: logout parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -330,7 +319,6 @@ paths: operationId: listRegistrationCountries description: 公开读接口，只返回 user-service 判定的 `enabled=true` 国家，并按 `sort_order,country_code` 排序。 parameters: - - $ref: "#/parameters/RequestIDHeader" responses: "200": description: 查询成功，`data.countries` 返回当前开放国家。 @@ -346,7 +334,6 @@ paths: operationId: getAppBootstrap description: 公开读接口，只返回版本、维护、feature flags、底部 tab 和配置版本号，不返回 banner、礼物或资源大列表。 parameters: - - $ref: "#/parameters/RequestIDHeader" responses: "200": description: 查询成功，`data` 返回 App 启动状态机需要的轻量配置摘要。 @@ -360,7 +347,6 @@ paths: operationId: listAppH5Links description: 公开读接口，返回后台管理系统 APP配置/H5配置 维护的 H5 地址。未配置 URL 的入口仍按固定 key 返回空字符串。 parameters: - - $ref: "#/parameters/RequestIDHeader" responses: "200": description: 查询成功，`data.items` 返回固定 H5 入口集合。 @@ -377,7 +363,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" responses: "200": description: 签发成功，`data.user_id` 使用内部长 `user_id` 的十进制字符串。 @@ -401,7 +386,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -442,7 +426,6 @@ paths: 腾讯 RTC 控制台配置的房间/媒体事件回调入口。gateway 校验 `SdkAppId` 与 `Sign` 后，只消费房间退出 `104`、开始推送音频 `203`、停止推送音频 `204`。 该接口不使用业务 envelope；成功或可忽略业务错误返回 `{"code":0}`。 parameters: - - $ref: "#/parameters/RequestIDHeader" - $ref: "#/parameters/TencentRTCSDKAppIDHeader" - $ref: "#/parameters/TencentRTCSignHeader" - name: app_code @@ -484,7 +467,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: file in: formData required: true @@ -513,7 +495,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: file in: formData required: true @@ -540,7 +521,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -568,7 +548,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -623,7 +602,6 @@ paths: summary: 通过当前展示短号解析用户 operationId: resolveDisplayUserID parameters: - - $ref: "#/parameters/RequestIDHeader" - $ref: "#/parameters/DisplayUserIDPath" responses: "200": @@ -648,7 +626,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: user_ids in: query required: true @@ -676,7 +653,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" responses: "200": description: 查询成功，`data` 返回当前用户短号状态。 @@ -700,7 +676,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" responses: "200": description: 查询成功，`data` 返回当前用户身份开关。 @@ -722,7 +697,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" responses: "200": description: 查询成功，`data` 返回我的页首屏渲染摘要。 @@ -748,7 +722,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -779,7 +752,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -810,7 +782,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -842,7 +813,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -874,7 +844,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -907,7 +876,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: tab in: query required: false @@ -956,7 +924,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: tab in: query required: true @@ -997,7 +964,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" responses: "200": description: 查询成功，`data.has_room=false` 表示当前用户尚未创建房间；有房间时 `data.room.im_group_id` 显式返回腾讯 IM 房间群 ID。 @@ -1019,7 +985,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" responses: "200": description: 查询成功，`data.has_current_room=false` 表示无需恢复。 @@ -1041,7 +1006,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: room_id in: query required: true @@ -1073,12 +1037,11 @@ paths: operationId: createRoom description: | gateway 从 access token 取当前 `user_id` 写入 room-service `RequestMeta.actor_user_id`，room-service 由该 actor 确定 owner 和默认 host。 - 请求体不接收 `room_id`、`command_id`、`owner_user_id` 或 `host_user_id`；gateway 生成内部 room_id 和 command_id。 + 请求体不接收 `room_id`、`owner_user_id` 或 `host_user_id`；`command_id` 由客户端按创建动作生成，gateway 生成内部 room_id。 同一个登录用户只能作为 owner 创建一个房间；已有房间时返回 409 Conflict。 security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1114,7 +1077,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1145,7 +1107,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1180,7 +1141,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1214,7 +1174,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1248,7 +1207,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1282,7 +1240,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1316,7 +1273,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1350,7 +1306,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1384,7 +1339,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1418,7 +1372,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1452,7 +1405,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1486,7 +1438,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1520,7 +1471,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1554,7 +1504,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1588,7 +1537,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1622,7 +1570,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1654,7 +1601,6 @@ paths: summary: App 资源列表 operationId: listResources parameters: - - $ref: "#/parameters/RequestIDHeader" - name: resource_type in: query required: false @@ -1699,7 +1645,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: resource_type in: query required: false @@ -1739,7 +1684,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -1772,7 +1716,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: scene in: query required: true @@ -1809,7 +1752,6 @@ paths: summary: App 资源组详情 operationId: getResourceGroup parameters: - - $ref: "#/parameters/RequestIDHeader" - $ref: "#/parameters/ResourceGroupIDPath" responses: "200": @@ -1829,7 +1771,6 @@ paths: summary: App 礼物列表 operationId: listGifts parameters: - - $ref: "#/parameters/RequestIDHeader" - name: keyword in: query required: false @@ -1868,7 +1809,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: resource_type in: query required: false @@ -1896,7 +1836,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - $ref: "#/parameters/ResourceIDPath" - name: body in: body @@ -1928,7 +1867,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" responses: "200": description: 查询成功。 @@ -1950,7 +1888,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: section in: query required: true @@ -1987,7 +1924,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -2015,7 +1951,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: message_id in: path required: true @@ -2043,7 +1978,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: message_id in: path required: true @@ -2070,7 +2004,6 @@ paths: security: - BearerAuth: [] parameters: - - $ref: "#/parameters/RequestIDHeader" - name: body in: body required: true @@ -3351,7 +3284,7 @@ definitions: properties: command_id: type: string - description: 客户端幂等命令 ID；同一业务重试必须复用。 + description: 客户端按用户动作生成的幂等命令 ID；同一动作重试必须复用。 target_user_id: type: string description: 目标用户 ID；gateway 会校验用户存在且 active。 @@ -3567,7 +3500,7 @@ definitions: properties: command_id: type: string - description: 钱包命令幂等键；客户端重试同一转账必须复用该值。 + description: 钱包命令幂等键；客户端按用户转账动作生成，同一动作重试必须复用。 target_user_id: type: integer format: int64 @@ -3622,9 +3555,13 @@ definitions: CreateRoomRequest: type: object required: + - command_id - mode - room_name properties: + command_id: + type: string + description: 房间创建命令幂等键；客户端按创建房间动作生成，同一动作的 HTTP 重试必须复用。 seat_count: type: integer format: int32 @@ -3657,6 +3594,7 @@ definitions: type: string command_id: type: string + description: 房间命令幂等键；客户端按用户动作生成，同一动作的 HTTP 重试必须复用。 room_name: type: string minLength: 1 @@ -3687,7 +3625,7 @@ definitions: type: string command_id: type: string - description: 房间命令幂等键；客户端重试同一业务动作必须复用该值。未传时 gateway 自动生成新值，仅保证本次请求可执行，不保证后续 HTTP 重试幂等。 + description: 房间命令幂等键；客户端按用户动作生成，同一动作的 HTTP 重试必须复用。 role: type: string RoomHeartbeatRequest: @@ -3699,7 +3637,7 @@ definitions: type: string command_id: type: string - description: 心跳命令幂等键；建议客户端每次心跳生成新值，HTTP 重试同一次心跳复用该值。 + description: 心跳命令幂等键；客户端按单次心跳动作生成，HTTP 重试同一次心跳必须复用。 LeaveRoomRequest: type: object properties: @@ -3707,7 +3645,7 @@ definitions: type: string command_id: type: string - description: 房间命令幂等键；客户端重试同一业务动作必须复用该值。未传时 gateway 自动生成新值，仅保证本次请求可执行，不保证后续 HTTP 重试幂等。 + description: 房间命令幂等键；客户端按用户动作生成，同一动作的 HTTP 重试必须复用。 MicUpRequest: type: object properties: @@ -3715,7 +3653,7 @@ definitions: type: string command_id: type: string - description: 房间命令幂等键；客户端重试同一业务动作必须复用该值。未传时 gateway 自动生成新值，仅保证本次请求可执行，不保证后续 HTTP 重试幂等。 + description: 房间命令幂等键；客户端按用户动作生成，同一动作的 HTTP 重试必须复用。 seat_no: type: integer format: int32 @@ -3726,7 +3664,7 @@ definitions: type: string command_id: type: string - description: 房间命令幂等键；客户端重试同一业务动作必须复用该值。未传时 gateway 自动生成新值，仅保证本次请求可执行，不保证后续 HTTP 重试幂等。 + description: 房间命令幂等键；客户端按用户动作生成，同一动作的 HTTP 重试必须复用。 target_user_id: type: integer format: int64 @@ -3740,7 +3678,7 @@ definitions: type: string command_id: type: string - description: 房间命令幂等键；客户端重试同一业务动作必须复用该值。未传时 gateway 自动生成新值，仅保证本次请求可执行，不保证后续 HTTP 重试幂等。 + description: 房间命令幂等键；客户端按用户动作生成，同一动作的 HTTP 重试必须复用。 target_user_id: type: integer format: int64 @@ -3759,7 +3697,7 @@ definitions: type: string command_id: type: string - description: 房间命令幂等键；客户端重试同一确认事件必须复用该值。 + description: 房间命令幂等键；客户端按单次确认事件生成，HTTP 重试同一确认事件必须复用。 target_user_id: type: integer format: int64 @@ -3782,7 +3720,7 @@ definitions: type: string command_id: type: string - description: 房间命令幂等键；客户端重试同一业务动作必须复用该值。未传时 gateway 自动生成新值，仅保证本次请求可执行，不保证后续 HTTP 重试幂等。 + description: 房间命令幂等键；客户端按用户动作生成，同一动作的 HTTP 重试必须复用。 seat_no: type: integer format: int32 @@ -3795,7 +3733,7 @@ definitions: type: string command_id: type: string - description: 房间命令幂等键；客户端重试同一业务动作必须复用该值。未传时 gateway 自动生成新值，仅保证本次请求可执行，不保证后续 HTTP 重试幂等。 + description: 房间命令幂等键；客户端按用户动作生成，同一动作的 HTTP 重试必须复用。 enabled: type: boolean SetRoomAdminRequest: @@ -3805,7 +3743,7 @@ definitions: type: string command_id: type: string - description: 房间命令幂等键；客户端重试同一业务动作必须复用该值。未传时 gateway 自动生成新值，仅保证本次请求可执行，不保证后续 HTTP 重试幂等。 + description: 房间命令幂等键；客户端按用户动作生成，同一动作的 HTTP 重试必须复用。 target_user_id: type: integer format: int64 @@ -3818,7 +3756,7 @@ definitions: type: string command_id: type: string - description: 房间命令幂等键；客户端重试同一业务动作必须复用该值。未传时 gateway 自动生成新值，仅保证本次请求可执行，不保证后续 HTTP 重试幂等。 + description: 房间命令幂等键；客户端按用户动作生成，同一动作的 HTTP 重试必须复用。 target_user_id: type: integer format: int64 @@ -3829,7 +3767,7 @@ definitions: type: string command_id: type: string - description: 房间命令幂等键；客户端重试同一业务动作必须复用该值。未传时 gateway 自动生成新值，仅保证本次请求可执行，不保证后续 HTTP 重试幂等。 + description: 房间命令幂等键；客户端按用户动作生成，同一动作的 HTTP 重试必须复用。 target_user_id: type: integer format: int64 @@ -3842,7 +3780,7 @@ definitions: type: string command_id: type: string - description: 房间命令幂等键；客户端重试同一业务动作必须复用该值。未传时 gateway 自动生成新值，仅保证本次请求可执行，不保证后续 HTTP 重试幂等。 + description: 房间命令幂等键；客户端按用户动作生成，同一动作的 HTTP 重试必须复用。 target_user_id: type: integer format: int64 @@ -3853,7 +3791,7 @@ definitions: type: string command_id: type: string - description: 房间命令幂等键；客户端重试同一业务动作必须复用该值。未传时 gateway 自动生成新值，仅保证本次请求可执行，不保证后续 HTTP 重试幂等。 + description: 房间命令幂等键；客户端按用户动作生成，同一动作的 HTTP 重试必须复用。 target_user_id: type: integer format: int64 @@ -3864,7 +3802,7 @@ definitions: type: string command_id: type: string - description: 房间命令幂等键；客户端重试同一业务动作必须复用该值。未传时 gateway 自动生成新值，仅保证本次请求可执行，不保证后续 HTTP 重试幂等。 + description: 房间命令幂等键；客户端按用户动作生成，同一动作的 HTTP 重试必须复用。 target_user_id: type: integer format: int64 diff --git a/docs/openapi/room.swagger.yaml b/docs/openapi/room.swagger.yaml index 89df1cb9..027b33ee 100644 --- a/docs/openapi/room.swagger.yaml +++ b/docs/openapi/room.swagger.yaml @@ -483,8 +483,10 @@ definitions: properties: request_id: type: string + description: 后端生成的链路追踪 ID，不参与幂等。 command_id: type: string + description: 客户端按用户写动作生成的幂等命令 ID，同一动作重试必须复用。 actor_user_id: type: integer format: int64 diff --git a/docs/主播公会BD架构.md b/docs/主播公会BD架构.md index 7ae633f4..a8f01cb8 100644 --- a/docs/主播公会BD架构.md +++ b/docs/主播公会BD架构.md @@ -530,7 +530,7 @@ POST /api/v1/bd-leader/invitations/agency POST /api/v1/wallet/coin-seller/transfer ``` -App API 必须使用 `/api/v1` envelope。用户身份来自 gateway 鉴权，`request_id` 只做链路追踪，不做幂等键；业务命令必须使用客户端生成或服务端下发的 `command_id`。 +App API 必须使用 `/api/v1` envelope。用户身份来自 gateway 鉴权，`request_id` 由后端生成且只做链路追踪，不做幂等键；业务命令必须使用客户端按用户动作生成的 `command_id`。 ### Internal gRPC Sketch diff --git a/docs/币商库存授信架构.md b/docs/币商库存授信架构.md index 149b17fd..ae45f19b 100644 --- a/docs/币商库存授信架构.md +++ b/docs/币商库存授信架构.md @@ -75,7 +75,7 @@ | `payment_ref` | 文本 | `usdt_purchase` 必填，建议填链上 tx hash、收款单号或财务凭证号 | | `evidence_ref` | 文本 | 凭证附件或后台上传后的文件 ID；首版建议两种类型都必填 | | `reason` | 文本 | 必填，写清楚进货或补偿原因 | -| `command_id` | 文本 | 前端生成或 admin-server 生成，用于幂等 | +| `command_id` | 文本 | 操作端前端按单次授信动作生成，同一动作重试必须复用 | 金额存储规则： @@ -419,4 +419,3 @@ USDT 进货金额属于财务口径，不能只放前端日志。 - 相同 `command_id` 但 payload 不同返回幂等冲突。 - wallet 事务失败时不写进货记录、不改余额。 - admin 审计记录包含 transaction_id、类型、金币数、USDT 金额和操作人。 - diff --git a/docs/房间Outbox补偿开发.md b/docs/房间Outbox补偿开发.md index 844a020b..67c41d6b 100644 --- a/docs/房间Outbox补偿开发.md +++ b/docs/房间Outbox补偿开发.md @@ -1,345 +1,144 @@ -# Room Outbox Compensation Worker Development Guide +# Room Outbox Worker -本文档定义 `room-service` 的 outbox 补偿 worker 如何接入进程生命周期、如何配置、如何重试，以及如何验证腾讯云 IM 同步失败后可由 `room_outbox` 补投。目标是收口 Phase 1 的房间系统消息可靠性，不扩展活动消费和 MQ 架构。 +本文档记录当前 `room-service` outbox 的实际语义。房间写命令只负责提交 Room Cell 状态、command log、snapshot 和 outbox；腾讯云 IM、activity-service 等外部投递全部由 outbox worker 异步执行。 -## Scope +## 边界 -本阶段只做： - -| Item | Result | +| 项目 | 当前规则 | | --- | --- | -| App 生命周期接入 | `room-service` 启动时启动 outbox worker，下线时先停止扫描新批次，再关闭 gRPC/MySQL/Redis | -| worker 配置 | 支持启停、扫描间隔、批量大小、单条投递超时和固定间隔重试策略 | -| 腾讯云 IM 补偿 | 同步 `PublishRoomEvent` 失败不回滚房间状态，pending outbox 后续补投到腾讯云 IM | -| 可验证测试 | 单测覆盖同步失败后状态已提交、outbox pending、worker 成功补投并标记 delivered | +| 主链路 | 不同步调用腾讯云 IM，不同步投递 activity-service | +| 持久化 | 命令变更、command log、room_outbox 在同一提交链路内完成 | +| worker | 读取 `room_outbox`，投递到配置的 outbox publisher fanout | +| 腾讯云 IM | `RoomCreated` 由 worker 确保群组存在，其他房间事件由 worker 发送群系统消息 | +| activity-service | 由 room outbox worker 异步投递，失败进入 room_outbox 重试 | +| 幂等 | 房间命令靠 `command_id`，outbox 投递靠 `event_id` | -本阶段不做： +不做的事： -| Not Included | Reason | +| 不做 | 原因 | | --- | --- | -| activity-service 消费主循环 | 活动消费需要独立消费位点，不能复用 IM 补偿状态 | -| MQ/Redis Stream | 当前 MySQL outbox 已能满足 Phase 1 补偿 | -| 精确一次投递 | 腾讯云 IM 和进程崩溃窗口只能保证 at-least-once，客户端按 `event_id` 去重 | -| 多 sink 状态表 | 先把腾讯云 IM 补偿闭环跑通；activity/audit 后续新增自己的 consumption 表 | -| 指数退避和 poison 队列 | 当前 `room_outbox` 没有 `next_retry_at_ms` 或 poison 状态，V1 用固定间隔重试 | +| 在 HTTP/gRPC 请求里等待 IM REST | 外部慢调用会拖住房间串行命令队列 | +| 在 Room Cell 中保存投递状态 | 投递状态属于 outbox，不属于房间权威状态 | +| 直接用 Redis 做 outbox | MySQL 是恢复和补偿的持久来源 | +| 精确一次投递 | 外部系统和进程崩溃窗口只能做到 at-least-once，消费端必须按 `event_id` 去重 | -## Current State - -当前代码已有基础能力： - -| Component | Current Fact | -| --- | --- | -| `room-service/internal/room/service/pipeline.go` | 命令成功后先写 command log/outbox，再尝试同步 `PublishRoomEvent`；同步成功 best-effort 标记 `delivered`，同步失败不回滚状态 | -| `room-service/internal/room/service/outbox_worker.go` | 已有 `RunOutboxWorker(ctx, options)` 持续循环和 `ProcessPendingOutbox(ctx, options)` 单批处理 | -| `room-service/internal/storage/mysql/repository.go` | 已有 `ClaimPendingOutbox`、`MarkOutboxDelivered`、`MarkOutboxFailed` | -| `room-service/internal/integration/tencent_im.go` | 已能把 `EventEnvelope` 转换成腾讯云 IM 房间系统事件 | -| `room-service/internal/app/app.go` | 已在进程生命周期中启动 presence stale worker 和 outbox worker，下线时先停 worker 再关闭 gRPC/MySQL/Redis | - -当前风险： - -- 代码接入前已经残留的 `pending` 事件仍会被补投；V1 接受 at-least-once，客户端按 `event_id` 去重。 -- `room_outbox.status` 统一为 `pending/delivering/delivered/retryable/failed`，不再使用旧的 `published` 命名。 -- 多个 room-service 实例通过 `ClaimPendingOutbox` 抢占同一批事件；未过期的 `delivering` 记录不会被其他 worker 重复抢占。 - -## Target Semantics - -房间命令成功后的语义： +## 写命令语义 ```text -Room Cell command applied - -> write command log - -> write room_outbox status=pending - -> persist snapshot if needed - -> replace in-memory RoomState - -> try synchronous PublishRoomEvent - -> if sync publish success, best-effort mark that event delivered - -> if sync publish fails, keep pending for worker +gateway -> room-service command + -> idempotency check + -> ensure Redis lease and Room Cell + -> mutate cloned RoomState + -> optional wallet debit before room gift state is committed + -> SaveMutation(command log + room_outbox + room meta) + -> snapshot when version hits snapshot_every_n + -> update list/presence projections best effort + -> return response + +outbox worker + -> claim pending/retryable rows + -> publish to composite outbox publisher + -> delivered / retryable / failed ``` -补偿 worker 语义： +主链路成功只表示房间状态已经落库；不表示腾讯云 IM 或 activity-service 已经完成投递。客户端实时表现应以房间响应、房间 snapshot 和 IM 系统消息共同补偿。 -```text -claim room_outbox where status in (pending, retryable) - -> status=delivering, set worker_id and lock_until_ms - -> publish one EventEnvelope to configured OutboxPublisher - -> success: status=delivered, last_error=NULL - -> failure: status=retryable, retry_count=retry_count+1, last_error= - -> next scan retries failed retryable records -``` +## 数据状态 -重要边界： +`room_outbox.status` 使用： -- worker 不执行房间命令，不修改 Room Cell，不写 command log，不写 snapshot。 -- worker 只处理已经持久化的 outbox envelope。 -- worker 投递失败不能影响 `room-service` ready；腾讯云 IM 抖动时 room 状态服务仍应接收房间命令。 -- `request_id` 仍然只做追踪；outbox 去重键是 `event_id`，房间命令幂等键是 `command_id`。 -- `delivered` 在本阶段表示腾讯云 IM 补偿通道已处理，不表示 activity/audit 已消费。 +| 状态 | 含义 | +| --- | --- | +| `pending` | 已落库，尚未被 worker 处理 | +| `delivering` | 已被 worker 抢占，`lock_until_ms` 前其他 worker 不抢 | +| `delivered` | 所有配置的 outbox publisher 已处理成功 | +| `retryable` | 上次投递失败，等待 `next_retry_at_ms` 后重试 | +| `failed` | 达到最大重试次数，进入死信，需要人工处理或后台工具重放 | -## Configuration +旧数据里可能存在历史 `published` 状态。当前 worker 不再写这个状态，新链路只写上表状态。 -在 `services/room-service/internal/config/config.go` 增加： +## 配置 -```go -type OutboxWorkerConfig struct { - Enabled bool `yaml:"enabled"` - PollInterval time.Duration `yaml:"poll_interval"` - BatchSize int `yaml:"batch_size"` - PublishTimeout time.Duration `yaml:"publish_timeout"` - RetryStrategy string `yaml:"retry_strategy"` -} -``` - -挂到 room-service 配置： - -```go -type Config struct { - // existing fields... - OutboxWorker OutboxWorkerConfig `yaml:"outbox_worker"` -} -``` - -默认值： - -| Field | Default | Meaning | -| --- | ---: | --- | -| `enabled` | `true` | 本地和生产默认启动补偿 worker；测试可显式关闭 | -| `poll_interval` | `1s` | 每轮扫描 pending outbox 的间隔 | -| `batch_size` | `100` | 每轮最多处理的 outbox 记录数 | -| `publish_timeout` | `3s` | 单条事件调用腾讯云 IM 的最大时间 | -| `retry_strategy` | `fixed_interval` | V1 只支持固定间隔重试，失败记录下一轮继续扫描 | - -配置文件示例： +`services/room-service/configs/*.yaml` 当前配置： ```yaml +snapshot_every_n: 10 + outbox_worker: enabled: true poll_interval: "1s" batch_size: 100 publish_timeout: "3s" - retry_strategy: "fixed_interval" + retry_strategy: "exponential_backoff" + max_retry_count: 10 + initial_backoff: "5s" + max_backoff: "5m" ``` 配置规则： -- `poll_interval <= 0` 时使用默认值，不能导致 busy loop。 -- `batch_size <= 0` 时使用默认值，不能全表扫描。 -- `publish_timeout <= 0` 时使用默认值，不能让单条外部调用无限阻塞 worker。 -- `retry_strategy` 只接受空值或 `fixed_interval`；未知值启动失败，避免配置写错后静默改变投递语义。 -- `config.yaml`、`config.docker.yaml`、`config.tencent.example.yaml` 必须同步增加该配置。 - -## App Lifecycle - -`room-service/internal/app.App` 需要新增 worker 生命周期字段： - -```go -type App struct { - // existing fields... - workerCancel context.CancelFunc - workerWG sync.WaitGroup -} -``` - -启动顺序： - -```text -1. 初始化 MySQL repository、Redis directory、wallet client、Tencent IM publisher -2. 创建 room service 和 gRPC server -3. Run 标记 gRPC serving -4. 创建 worker context -5. 启动 presence stale worker -6. 如果 outbox_worker.enabled=true，启动 outbox worker -7. 进入 grpcServer.Serve -``` - -关闭顺序： - -```text -1. MarkDraining -2. cancel worker context，停止扫描新 outbox 批次 -3. worker 当前正在处理的单条事件受 publish_timeout 控制，结束后退出 -4. workerWG.Wait 等待后台 worker 停止 -5. GracefulStop gRPC，等待已进入命令完成持久化 -6. 关闭 wallet gRPC、MySQL、Redis -``` - -关闭规则： - -- 不能在 worker 退出前关闭 MySQL，否则可能无法标记投递结果。 -- worker 收到 cancel 后不再拉取新批次；已经进入投递的事件可以完成成功标记或失败标记。 -- 如果进程被强杀，已抢占未完成的记录保持 `delivering`，`lock_until_ms` 过期后下次启动继续补偿。 - -## Worker API - -在 `room/service` 增加持续循环方法： - -```go -type OutboxWorkerOptions struct { - PollInterval time.Duration - BatchSize int - PublishTimeout time.Duration - RetryStrategy string -} - -func (s *Service) RunOutboxWorker(ctx context.Context, options OutboxWorkerOptions) -``` - -`RunOutboxWorker` 只负责循环、节流和退出： - -```text -normalize options -process one batch immediately -start ticker -for each tick: - if ctx done: return - ProcessPendingOutbox(ctx, options) -``` - -`ProcessPendingOutbox` 建议改成 options 版本： - -```go -func (s *Service) ProcessPendingOutbox(ctx context.Context, options OutboxWorkerOptions) error -``` - -处理规则： - -- 每轮调用 `repository.ClaimPendingOutbox(ctx, worker_id, batch_size, lock_until_ms)`，把 `pending/retryable` 事件抢占为 `delivering`。 -- 每条记录单独创建 `context.WithTimeout(ctx, publish_timeout)`。 -- `PublishOutboxEvent` 成功后调用 `MarkOutboxDelivered`。 -- `PublishOutboxEvent` 失败后调用 `MarkOutboxFailed`。 -- 单条失败不阻断同批后续记录。 -- `ClaimPendingOutbox` 或 `MarkOutboxDelivered/Failed` 失败属于本轮 worker 错误，记录后等待下一轮。 - -## Sync Publish Interaction - -为了避免 worker 对同步成功事件重复补投，命令链路需要在同步成功后 best-effort 标记 outbox： - -```text -if result.applied && result.syncEvent != nil: - err := syncPublisher.PublishRoomEvent(ctx, *result.syncEvent) - if err == nil: - repository.MarkOutboxDelivered(ctx, result.syncEvent.EventID) - if err != nil: - keep room_outbox pending -``` - -约束： - -- `MarkOutboxDelivered` 失败不能让已经成功提交的房间命令返回失败；最多造成后续重复投递。 -- 同步成功但进程在标记 delivered 前崩溃时，worker 后续可能重复投递；客户端必须用 `event_id` 去重。 -- 对 `RoomCreated`、`RoomHeatChanged`、`RoomRankChanged` 这类不推 IM 的事件，`PublishOutboxEvent` 返回 no-op success 后可标记 `delivered`，避免长期 pending。 - -## Retry Strategy - -V1 采用固定间隔无限重试： - -| Case | Behavior | +| 字段 | 规则 | | --- | --- | -| 腾讯云 IM 网络错误 | `retry_count + 1`，转为 `retryable`，下一轮继续 | -| 腾讯云 IM 返回可重试错误 | `retry_count + 1`，转为 `retryable`，下一轮继续 | -| envelope 反序列化失败 | worker 本轮返回错误，不删除记录；后续需要人工排查或补工具 | -| 不需要推 IM 的事件类型 | no-op success，标记 `delivered` | -| 进程退出 | 已抢占未完成的记录保持 `delivering`，`lock_until_ms` 过期后可被重新抢占 | +| `snapshot_every_n` | 开发和生产默认 `10`；命令日志仍保证恢复，不能依赖每条命令快照 | +| `poll_interval` | worker 空轮询间隔，必须大于 0，避免 busy loop | +| `batch_size` | 每轮最大抢占条数，必须大于 0，避免全表扫描 | +| `publish_timeout` | 单条 outbox 投递 deadline，避免外部系统拖死 worker | +| `retry_strategy` | 当前为 `exponential_backoff` | +| `max_retry_count` | 达到次数后标记 `failed`，防止 poison event 永久污染日志 | +| `initial_backoff` | 第一次失败后的最小退避 | +| `max_backoff` | 指数退避上限 | -为什么不加最大重试次数： +## 重试和死信 -- 房间系统消息是状态变更的外部表现，不能因为达到次数就丢弃。 -- 当前 schema 没有 poison 状态，贸然新增半成品状态会让恢复和运营查询复杂化。 -- `retry_count` 和 `last_error` 已足够驱动告警；后续需要 poison 队列时再单独设计 `failed/next_retry_at_ms`。 - -## Observability - -首版至少打结构化日志字段： - -| Field | Meaning | -| --- | --- | -| `worker` | 固定为 `room_outbox` | -| `node_id` | 当前 room-service 节点 | -| `event_id` | outbox 事件 ID | -| `event_type` | RoomUserJoined、RoomGiftSent 等 | -| `room_id` | 事件所属房间 | -| `retry_count` | 当前失败次数 | -| `status` | pending、delivering、delivered、retryable 或 failed | -| `error` | 最近一次失败原因，不能包含密钥或完整腾讯响应体 | - -指标后续可补： +失败处理： ```text -room_outbox_pending_total -room_outbox_publish_success_total -room_outbox_publish_failure_total -room_outbox_publish_latency_ms -room_outbox_oldest_pending_age_ms +publish failed + -> retry_count + 1 + -> if retry_count >= max_retry_count: status=failed + -> else: status=retryable, next_retry_at_ms=now+backoff ``` -## Implementation Order +`failed` 不再自动重试。后台应提供人工查看和重放工具，重放时必须保留原 `event_id` 或生成明确的新修复事件，避免消费端无法去重。 -按以下顺序做，避免 worker 一启动就制造重复消息： +## 投递对象 -1. 扩展 room-service 配置结构和三份 YAML，增加 `outbox_worker` 配置。 -2. 给 `room/service` 增加 `OutboxWorkerOptions` 和 `RunOutboxWorker`。 -3. 调整 `ProcessPendingOutbox`，支持 `batch_size`、单条 `publish_timeout`、单条失败不中断批次。 -4. 调整命令链路：同步 `PublishRoomEvent` 成功后 best-effort `MarkOutboxDelivered(event_id)`；同步失败保留初始 `pending`，worker 投递失败转为 `retryable`。 -5. 在 `app.Run` 启动 outbox worker，在 `app.Close` cancel 并等待 worker 退出。 -6. 增加单测覆盖同步失败补偿、重试失败转为 `retryable`、worker cancel 退出。 -7. 更新 README 当前限制，说明 outbox worker 已接入生命周期。 +当前 `CompositeOutboxPublisher` fanout 到多个 publisher： -## Acceptance Tests - -必须覆盖： - -| Test | Expected Result | +| Publisher | 行为 | | --- | --- | -| 同步 IM 失败后命令成功 | `JoinRoom` 或 `SendGift` 返回成功，RoomSnapshot 已更新 | -| 同步 IM 失败后 outbox pending | 对应 `event_id` 仍为 `pending`，`retry_count` 未因同步失败被提前增加 | -| worker 补投成功 | fake outbox publisher 收到同一 `event_id`，repository 标记 `delivered` | -| worker 补投失败 | 记录转为 `retryable`，`retry_count+1`，`last_error` 有值 | -| 下一轮成功 | 第一次失败、第二次成功后状态变为 `delivered` | -| 同步 IM 成功 | 对应 `event_id` best-effort 标记 `delivered`，worker 不再扫描该事件 | -| worker cancel | cancel 后不再扫描新批次，当前投递受 `publish_timeout` 约束后退出 | -| 非 IM 事件 | `RoomCreated/RoomHeatChanged/RoomRankChanged` 不发送腾讯 IM，但会标记 `delivered` | +| Tencent IM | `RoomCreated` 确保群存在；房间系统事件发送腾讯云 IM 群自定义消息 | +| Activity | 将 room outbox 事件投递给 `activity-service` 的消费入口 | -已有测试 `TestRoomOutboxCompensationWhenSyncPublishFails` 可以作为基础，但需要补齐持续 worker 和同步成功去重场景。 +fanout 要求所有 publisher 成功后才标记 `delivered`。任一失败都会进入重试，因此消费端必须幂等处理重复事件。 -## Verification Commands +## 阶段耗时日志 -代码完成后至少运行： +每条房间写命令都会输出 `room_command_timing`： -```bash -go test ./services/room-service/internal/room/service -go test ./services/room-service/internal/config -go test ./services/room-service/internal/app -go test ./... -``` +| 字段 | 含义 | +| --- | --- | +| `idempotency_ms` | command log 幂等检查耗时 | +| `lease_ms` | Redis lease / Room Cell 装载耗时 | +| `wallet_debit_ms` | 送礼扣费耗时；非送礼为 0 | +| `save_mutation_ms` | command log + outbox + 房间 meta 提交耗时 | +| `snapshot_ms` | 快照持久化耗时；未触发为 0 | +| `projection_ms` | 列表和 presence 投影耗时 | +| `activity_ms` | 主链路 activity 同步耗时；当前固定为 0 | +| `tencent_im_ms` | 主链路腾讯 IM 同步耗时；当前固定为 0 | +| `activity_async_outbox` | 本命令是否产生异步 activity/outbox 投递 | +| `tencent_im_async_outbox` | 本命令是否产生异步 IM/outbox 投递 | -涉及配置和生命周期后运行： +排查接口卡顿时先看 `total_ms`，再看各阶段字段。送礼慢通常优先看 `wallet_debit_ms`，普通房间命令慢优先看 `save_mutation_ms`、`snapshot_ms` 和 `projection_ms`。 -```bash -docker compose config -docker compose build room-service -``` +## 验收 -如果要验证本地启动链路： +必须满足： -```bash -docker compose up -d mysql redis wallet-service room-service -docker compose ps -docker compose down -``` - -## Done Definition - -完成标准： - -```text -room command committed - -> outbox persisted - -> sync Tencent IM publish fails - -> HTTP/gRPC command response still succeeds - -> outbox remains pending - -> app-level worker scans pending - -> worker publishes same event_id to Tencent IM publisher - -> outbox status becomes delivered - -> process shutdown stops worker before closing MySQL -``` - -该闭环稳定后，Phase 1 的“房间系统消息同步失败后可补投”才算完成。 +| 场景 | 验收 | +| --- | --- | +| IM 不可用 | 房间写命令仍成功，outbox 进入 `retryable` 或最终 `failed` | +| activity 不可用 | 房间写命令仍成功，outbox 进入重试 | +| poison event | 达到 `max_retry_count` 后进入 `failed`，不再每秒刷日志 | +| 正常投递 | 同房间事件最终全部 `delivered` | +| 真实链路 | create/join/mic/sendGift/leave 均有 `room_command_timing` | diff --git a/docs/接口清单.md b/docs/接口清单.md index 78007cc5..97193ea6 100644 --- a/docs/接口清单.md +++ b/docs/接口清单.md @@ -142,12 +142,12 @@ | PUT | `/api/v1/admin/gifts/{gift_id}` | admin | `updateGift` | updateGift | | POST | `/api/v1/admin/gifts/{gift_id}/disable` | admin | `disableGift` | disableGift | | POST | `/api/v1/admin/gifts/{gift_id}/enable` | admin | `enableGift` | enableGift | -| GET | `/api/v1/admin/game/platforms` | admin | `listGamePlatforms` | 查询游戏平台配置 | -| POST | `/api/v1/admin/game/platforms` | admin | `upsertGamePlatform` | 创建或更新游戏平台 | -| PATCH | `/api/v1/admin/game/platforms/{platform_code}` | admin | `updateGamePlatform` | 更新游戏平台 | -| GET | `/api/v1/admin/game/games` | admin | `listGameCatalog` | 查询游戏目录 | -| POST | `/api/v1/admin/game/games` | admin | `upsertGameCatalog` | 创建或更新游戏目录 | -| PATCH | `/api/v1/admin/game/games/{game_id}` | admin | `updateGameCatalog` | 更新游戏目录 | +| GET | `/api/v1/admin/game/platforms` | admin | `listPlatforms` | 查询游戏平台配置 | +| POST | `/api/v1/admin/game/platforms` | admin | `createPlatform` | 创建游戏平台 | +| PATCH | `/api/v1/admin/game/platforms/{platform_code}` | admin | `updatePlatform` | 更新游戏平台 | +| GET | `/api/v1/admin/game/games` | admin | `listCatalog` | 查询游戏目录 | +| POST | `/api/v1/admin/game/games` | admin | `createCatalog` | 创建游戏目录 | +| PATCH | `/api/v1/admin/game/games/{game_id}` | admin | `updateCatalog` | 更新游戏目录 | | PATCH | `/api/v1/admin/game/games/{game_id}/status` | admin | `setGameStatus` | 设置游戏状态 | | GET | `/api/v1/admin/hosts` | admin | `listHosts` | listHosts | | GET | `/api/v1/admin/payment/recharge-bills` | admin | `listRechargeBills` | listRechargeBills | diff --git a/docs/服务基础技术设计.md b/docs/服务基础技术设计.md index cef07b75..f85de1a5 100644 --- a/docs/服务基础技术设计.md +++ b/docs/服务基础技术设计.md @@ -462,7 +462,7 @@ activity_outbox gateway 接入这些服务时只做： - HTTP JSON 入参解析。 -- 生成或透传 `request_id`。 +- 生成 `request_id` 并透传到内部调用。 - 调内部 gRPC。 - 把 gRPC status + reason 映射为 HTTP envelope。 diff --git a/docs/每日任务系统架构.md b/docs/每日任务系统架构.md index 22b10985..2e905ca7 100644 --- a/docs/每日任务系统架构.md +++ b/docs/每日任务系统架构.md @@ -194,7 +194,7 @@ POST /api/v1/tasks/claim "task_id": "123", "task_type": "daily", "task_day": "2026-05-09", - "command_id": "client-or-gateway-generated-id" + "command_id": "cmd_task_claim_550e8400" } ``` @@ -204,7 +204,7 @@ POST /api/v1/tasks/claim - 必须未领取。 - daily 只能领取当前 `task_day` 的任务。 - exclusive 只能领取 `lifetime` 任务。 -- `command_id` 幂等，重复请求返回同一结果。 +- `command_id` 由客户端按领取动作生成，同一次动作重试必须复用，重复请求返回同一结果。 可选第二阶段增加： diff --git a/docs/游戏接入架构.md b/docs/游戏接入架构.md index fee89c12..78503bfa 100644 --- a/docs/游戏接入架构.md +++ b/docs/游戏接入架构.md @@ -444,7 +444,7 @@ POST /api/v1/game-callbacks/{platform_code}/{operation} gateway 行为： - 保留原始 body、query、header，不提前解析成统一 JSON。 -- 生成或透传 `request_id`。 +- 生成 `request_id` 并透传到内部调用。 - 按 `platform_code` 转发给 game-service。 - 回调响应不套 App 业务 envelope，因为第三方通常要求固定响应格式。 diff --git a/docs/登录注册技术设计.md b/docs/登录注册技术设计.md index ee360f07..8f96a2bb 100644 --- a/docs/登录注册技术设计.md +++ b/docs/登录注册技术设计.md @@ -65,7 +65,7 @@ v1 不支持： } ``` -`request_id` 来自 `X-Request-ID` 或由 gateway 生成，并透传到内部 gRPC meta。 +`request_id` 由 gateway 生成，并透传到内部 gRPC meta；客户端不需要提交 `X-Request-ID`。 ### Third Party Login Or Register diff --git a/docs/语音房功能清单.md b/docs/语音房功能清单.md index 0ada834b..f46474d2 100644 --- a/docs/语音房功能清单.md +++ b/docs/语音房功能清单.md @@ -19,7 +19,7 @@ | 长连接和房间消息 | 腾讯云 IM 负责客户端长连接、群消息、公屏、离线/漫游和全球接入 | `DONE` | | 真实语音 | 腾讯云 RTC 负责音频频道和音频流 | `PARTIAL` | | 账务 | `wallet-service` 负责扣费语义，`room-service.SendGift` 扣费成功后才落房间表现 | `DONE` | -| 活动 | `activity-service` 应消费 room outbox，不侵入 Room Cell 主链路 | `PARTIAL` | +| 活动 | `activity-service` 只通过 room outbox 异步接收房间事件，不侵入 Room Cell 主链路 | `PARTIAL` | | 用户资料 | `user-service` 负责用户主数据、短号、头像、国家等资料 | `DONE` | ## Foundation And Identity @@ -44,8 +44,8 @@ | 房间详情页聚合接口 | `DONE` | `/rooms/{room_id}/detail` 返回 room/viewer/seats/rank_top/online_count/permissions/im/rtc | 头像框、座驾、VIP、等级等展示 read model 仍需接入 | | 房间在线用户分页 | `DONE` | `/rooms/{room_id}/online-users` 读取 `room_user_presence` 分页投影 | 大页码仍是 offset 分页，超大房间可升级 cursor | | 房间礼物面板 | `DONE` | `/rooms/{room_id}/gift-panel` 聚合 COIN 余额、收礼人、礼物 tabs、礼物配置和数量预设 | all_mic/all_room/couple 账务拆单策略未开放 | -| 房间命令 `command_id` | `DONE` | 创建房间由 gateway 生成内部 `command_id`；其他写命令支持 body `command_id`，未传时 gateway 生成一次性值 | 客户端只在非创建类重试动作复用 command_id | -| `request_id` 追踪 | `DONE` | request_id 透传 RequestMeta，不作为幂等键 | 日志和链路追踪系统可继续完善 | +| 房间命令 `command_id` | `DONE` | 写命令使用前端按用户动作生成的 `command_id` 做服务端幂等 | 同一次用户动作的 HTTP 重试必须复用 command_id | +| `request_id` 追踪 | `DONE` | gateway 生成 request_id 并透传 RequestMeta，不作为幂等键 | 日志和链路追踪系统可继续完善 | | IM UserSig 签发 | `DONE` | `/api/v1/im/usersig` 使用 access token `user_id` 签发 | 真实腾讯云应用联调和告警 | | RTC token 签发 | `DONE` | `JoinRoom` 响应内嵌首屏 RTC token；`/api/v1/rtc/token` 保留为过期刷新和补偿入口 | 客户端 TRTC SDK 实际进房联调 | | 腾讯 IM 回调入口 | `DONE` | `/api/v1/tencent-im/callback` 支持入群前和发言前守卫 | 腾讯云控制台回调 URL 和鉴权参数真实配置 | @@ -55,7 +55,7 @@ | Feature | Status | Current Evidence | Remaining Work | | --- | --- | --- | --- | -| 创建房间 | `DONE` | `CreateRoom` 校验 room_id、actor、mode；`seat_count=0` 使用后台默认值，正数必须命中后台启用配置；写 meta/command/outbox/snapshot | 真实 IM 建群失败注入和线上重试策略联调 | +| 创建房间 | `DONE` | `CreateRoom` 校验 room_id、actor、mode；`seat_count=0` 使用后台默认值，正数必须命中后台启用配置；写 meta/command/outbox/snapshot 后返回 | 真实 IM 建群由 outbox worker 异步完成，客户端 IM joinGroup 需要容忍短暂 group-not-ready 并重试 | | 修改房间资料和座位数 | `DONE` | `UpdateRoomProfile` 由房间内 owner 执行；资料和座位数变更经过 Room Cell，缩容要求被删除麦位空闲且未锁 | App 编辑页和真实 IM `room_profile_updated` 展示联调 | | owner/host 收敛 | `DONE` | room-service 从 `actor_user_id` 默认 owner/host，拒绝外部冒用 | 后续转主持需独立命令 | | 进房 presence | `DONE` | `JoinRoom` 写 `OnlineUsers`，gateway 返回首屏房间 DTO、`im_group_id` 和 RTC token | 客户端断线重连自动化测试 | @@ -74,21 +74,21 @@ | --- | --- | --- | --- | | MySQL 持久化 | `DONE` | `rooms`、`room_snapshots`、`room_command_log`、`room_outbox` | 线上迁移脚本和容量评估 | | Redis room lease | `DONE` | Redis directory 做 room owner lease，保持单房间单活 | lease 续约/释放策略需要真实故障演练 | -| command log 幂等 | `DONE` | `HasCommand` 检查 command_id，命令重复返回当前快照 | 客户端 command_id 规范需要写入客户端协议 | -| snapshot 频率 | `DONE` | `snapshot_every_n` 配置，默认每次保存 | 高并发房间可能要调大并验证恢复成本 | +| command log 幂等 | `DONE` | `HasCommand` 检查 command_id，命令重复返回当前快照 | 客户端必须按用户动作生成并在重试时复用 command_id | +| snapshot 频率 | `DONE` | `snapshot_every_n` 配置，本地/生产默认 `10`，命令日志保证恢复 | 高并发房间可按恢复成本调到 20 并压测 | | outbox 持久化 | `DONE` | 房间事件写 `room_outbox` | 事件消费侧 schema 演进策略需补 | | outbox worker 生命周期 | `DONE` | app 生命周期启动/停止 room outbox worker | 外部消费者不止腾讯 IM 时需要 fanout 策略 | -| outbox 重试配置 | `DONE` | poll interval、batch size、publish timeout、fixed_interval | 指数退避、死信和告警未做 | +| outbox 重试配置 | `DONE` | poll interval、batch size、publish timeout、指数退避、最大重试和死信 | 后台死信查看、人工重放和告警规则需补 | ## Tencent IM Integration | Feature | Status | Current Evidence | Remaining Work | | --- | --- | --- | --- | | 客户端 IM UserSig | `DONE` | gateway 签发，identifier 使用内部 `user_id` 十进制字符串 | 线上密钥管理和过期刷新体验 | -| room-service 建群 | `DONE` | `EnsureRoomGroup` 调腾讯 REST create_group | 腾讯云真实错误码覆盖和已存在群语义验证 | +| room-service 建群 | `DONE` | `RoomCreated` outbox 由 worker 调腾讯 REST create_group | 腾讯云真实错误码覆盖和已存在群语义验证 | | 房间系统消息 | `DONE` | `PublishRoomEvent` 发送 TIMCustomElem，事件包含 room_version | 客户端消息解析和版本去重 | -| 同步推失败不回滚 | `DONE` | 命令提交后同步 publish 失败保持 outbox pending | 需要监控 pending 积压 | -| outbox 补投 IM | `DONE` | outbox worker 后续 publish 并标记 published | 多消费者场景需要事件状态拆分 | +| 主链路不阻塞 IM | `DONE` | 命令提交 command log/outbox 后返回；腾讯 IM 只由 outbox worker 异步投递 | 需要监控 pending/retryable/failed 积压 | +| outbox 投递 IM | `DONE` | outbox worker publish 成功后标记 `delivered`，失败指数退避，超限进 `failed` | 多消费者场景需要更细粒度 sink 状态 | | 入群前守卫 | `DONE` | `Group.CallbackBeforeApplyJoinGroup` -> `VerifyRoomPresence` | 腾讯云控制台配置和公网回调验收 | | 发言前守卫 | `DONE` | `Group.CallbackBeforeSendMsg` -> `CheckSpeakPermission` | 公屏消息内容审核、频控未做 | | 群历史补拉 | `PARTIAL` | 设计上依赖腾讯云 IM 群历史 | 客户端补拉窗口、服务端兜底 API 未实现 | @@ -135,14 +135,14 @@ | 本地礼物榜 | `DONE` | `LocalRank` 和 `gift_rank` 快照 | 全局榜、跨房榜、榜单重置未做 | | 礼物配置 | `PARTIAL` | `wallet_gift_prices` 提供服务端价格和 `COIN`/`DIAMOND` 收费资产，`gift_configs` 提供礼物类型、有效期和特效；SendGift 请求只传 `gift_id/gift_count`；礼物配置已绑定 `resource_type=gift` 资源 | 客户端礼物表现联调、资源版本缓存刷新未做 | | 背包/道具 | `PARTIAL` | `user_resource_entitlements`、`user_resource_equipment` 支持资源发放、我的资源和可佩戴资源装备 | 背包扣减、免费礼物、道具过期扫描未做 | -| 礼物消息补偿 | `DONE` | GiftSent/Heat/Rank 事件进入 outbox，主事件同步 IM | 客户端去重和展示策略未验收 | +| 礼物消息补偿 | `DONE` | GiftSent/Heat/Rank 事件进入 outbox，由 worker 异步投 IM/activity | 客户端去重和展示策略未验收 | ## Activity And Lucky Gift | Feature | Status | Current Evidence | Remaining Work | | --- | --- | --- | --- | -| activity-service 底座 | `PARTIAL` | gRPC Ping/GetActivityStatus、MySQL activity/outbox 表存在 | 尚未消费 room outbox | -| room outbox 消费边界 | `PARTIAL` | `consumer/room` 有接口骨架 | 真实 polling/MQ/offset/idempotency 未实现 | +| activity-service 底座 | `PARTIAL` | gRPC Ping/GetActivityStatus、MySQL activity/outbox 表存在 | 具体活动规则和奖励结算未实现 | +| room outbox 消费边界 | `DONE` | room outbox worker 异步投递 activity-service，失败跟随 room_outbox 重试/死信 | activity 内部消费幂等表和后台排障页需补 | | 幸运礼物开发文档 | `DONE` | `docs/幸运礼物Activity服务开发.md` | 按文档实现 proto、pool、risk、draw | | 活动配置 | `TODO` | 只有 activity status projection | 活动规则、时间、奖池、风控配置未实现 | | LuckyGift Check | `TODO` | 文档有设计，proto/code 未实现 | `CheckLuckyGift` RPC 和规则校验 | @@ -184,14 +184,14 @@ | 基础封禁 | `PARTIAL` | room 内 Kick 写 ban，user-service 有用户状态 | 全局封禁、设备/IP 黑名单、后台操作未做 | | 刷屏风控 | `TODO` | 当前只有 IM 发言权限守卫 | 需要消息频控、敏感词、内容安全 | | 房间后台 | `TODO` | 当前无 admin API | 查房、关房、踢人、禁言、封禁、改配置 | -| 指标和告警 | `PARTIAL` | health 和日志存在，文档提到指标 | outbox 积压、IM/RTC 错误率、房间命令耗时、wallet 失败率未完整实现 | +| 指标和告警 | `PARTIAL` | health、结构化日志和 `room_command_timing` 已存在 | outbox 积压、IM/RTC 错误率、wallet 失败率告警未完整实现 | ## Client Required Work | Feature | Status | Server Dependency | Remaining Work | | --- | --- | --- | --- | | 登录/注册 | `PARTIAL` | gateway/user-service 已有接口 | 客户端提交 Firebase ID token；服务端接 Firebase verifier 后形成生产登录闭环 | -| CreateRoom/JoinRoom/LeaveRoom | `PARTIAL` | gateway/room-service 已有接口 | 非创建类 command_id、重试、状态覆盖 | +| CreateRoom/JoinRoom/LeaveRoom | `PARTIAL` | gateway/room-service 已有接口 | command_id、重试和状态覆盖联调 | | 腾讯 IM 登录和进群 | `PARTIAL` | `/api/v1/im/usersig` 和回调守卫已存在 | SDK login、joinGroup、回调配置、历史补拉 | | 腾讯 RTC 进房 | `PARTIAL` | `/api/v1/rtc/token` 已存在 | TRTC enterRoom、角色切换、断线重连 | | 麦位 UI | `PARTIAL` | mic commands 已存在 | 上麦后推流、下麦停止推流、锁麦状态展示 | @@ -254,11 +254,11 @@ 当前服务端已经完成或基本完成： - 用户注册登录、短号、资料和国家修改底座。 -- gateway `/api/v1` envelope、房间命令 HTTP 路由、创建房间内部 `command_id` 生成和非创建命令 `command_id` 透传。 +- gateway `/api/v1` envelope、服务端 `request_id` 生成、房间命令 HTTP 路由和写命令 `command_id` 透传。 - room-service Room Cell、Create/Join/Leave、麦位上下/换位、禁言、踢人、SendGift。 - MySQL command log/snapshot/outbox、Redis lease、恢复和 stale presence 清理。 - 腾讯 IM UserSig、建群、系统消息、入群/发言回调守卫。 -- room outbox worker 补偿投递。 +- room outbox worker 异步投递腾讯 IM/activity，支持指数退避和死信。 - wallet-service DebitGift 幂等扣费。 - gateway RTC token 签发和 `user_id/room_id` 映射。 diff --git a/docs/语音房基础闭环实现.md b/docs/语音房基础闭环实现.md index dfeb7e40..61de7060 100644 --- a/docs/语音房基础闭环实现.md +++ b/docs/语音房基础闭环实现.md @@ -18,8 +18,8 @@ docs/房间Cell状态机.md | HTTP 命令入口 | `gateway-service` | 所有房间命令走 `/api/v1` envelope | | 房间业务态 | `room-service` | Room Cell 持有 presence、麦位、禁言、踢人、快照和恢复 | | 腾讯云 IM 登录票据 | `gateway-service` | `/api/v1/im/usersig` 返回 `sdk_app_id/user_id/user_sig/expire_at_ms` | -| 腾讯云 IM 群组 | `room-service` | `CreateRoom` 时确保 `room_id` 对应群组存在 | -| 房间系统消息 | `room-service` | 房间事件同步推腾讯云 IM，失败后由 outbox 补投 | +| 腾讯云 IM 群组 | `room-service` | `RoomCreated` outbox 由 worker 异步确保 `room_id` 对应群组存在 | +| 房间系统消息 | `room-service` | 房间事件只写 outbox，腾讯云 IM 由 outbox worker 异步投递 | | 实时权限守卫 | `gateway-service` + `room-service` | 腾讯云 IM 入群前和群内发言前必须回查房间业务态 | | 腾讯云 RTC 进房票据 | `gateway-service` | `JoinRoom` 成功后在响应中内嵌 RTC token；`/api/v1/rtc/token` 只做补偿和刷新 | @@ -85,10 +85,10 @@ sequenceDiagram C->>G: POST /api/v1/rooms/create(room_name, room_avatar?, room_description?) G->>R: CreateRoom(actor_user_id, room profile) - R->>T: REST create_group(room_id) R->>R: command log + snapshot + outbox R-->>G: RoomSnapshot G-->>C: room envelope + R-->>T: outbox worker create_group(room_id) C->>G: GET /api/v1/im/usersig G-->>C: SDKAppID + user_id + UserSig @@ -97,9 +97,9 @@ sequenceDiagram C->>G: POST /api/v1/rooms/join G->>R: JoinRoom(actor_user_id) R->>R: update Room Cell presence - R->>T: REST send_group_msg(room_user_joined) R-->>G: RoomSnapshot G-->>C: initial room envelope + im_group_id + RTC token + R-->>T: outbox worker send_group_msg(room_user_joined) C->>T: joinGroup(room_id) T->>G: CallbackBeforeApplyJoinGroup @@ -177,7 +177,7 @@ gateway 不保存房间状态，只做回调鉴权、字段解析、调用 room ### 1. CreateRoom -`CreateRoom` 必须在房间创建链路中确保腾讯云 IM 群组存在。 +`CreateRoom` 不等待腾讯云 IM 建群。它只提交房间权威状态和 `RoomCreated` outbox；建群由 outbox worker 异步完成。 同一个 `actor_user_id` 只能作为 owner 创建一个房间；加入、主持或管理别人的房间不受该 owner 限制影响。 主链路顺序： @@ -187,17 +187,17 @@ validate request acquire Redis lease check room meta check owner has no room -EnsureRoomGroup(room_id) initialize RoomState write room meta + command log + snapshot + outbox install Room Cell return snapshot ``` -如果 `EnsureRoomGroup` 失败： +如果后续 `EnsureRoomGroup` 失败： -- `CreateRoom` 不应该提交房间状态。 -- 客户端可以重试同一个创建请求。 +- `CreateRoom` 已经成功，不回滚房间状态。 +- outbox worker 按指数退避重试，达到上限进入 `failed`。 +- 客户端 IM SDK 加群可能短暂遇到 group-not-ready，需要按产品可接受窗口重试。 - 失败原因不应该泄露腾讯云密钥、UserSig 或完整公网响应。 ### 2. JoinRoom @@ -221,7 +221,7 @@ return snapshot - 用户不在房间内时幂等返回，不报错。 - 用户离房后 `VerifyRoomPresence` 必须返回 `present=false`。 - 如果用户在麦上，离房必须同步释放麦位。 -- 离房事件写 outbox，并尝试推腾讯云 IM 系统消息。 +- 离房事件写 outbox，腾讯云 IM 系统消息由 worker 异步投递。 ### 4. Guard RPC @@ -319,7 +319,7 @@ room-service 发给腾讯云 IM 的房间系统消息使用 `TIMCustomElem`。 4. **补 JoinRoom 重连刷新**：重复 JoinRoom 刷新 `last_seen_at_ms`，但不重复发进房系统消息。 5. **补 LeaveRoom 麦位释放**：离房时如果在麦上，释放麦位并产生一致的房间事件。 6. **补 presence stale worker**：按 `last_seen_at_ms` 清理超时 presence，写 `RoomUserLeft`。 -7. **补集成测试**：覆盖 UserSig、CreateRoom 建群、JoinRoom、回调守卫、LeaveRoom、重连恢复。 +7. **补集成测试**：覆盖 UserSig、CreateRoom outbox 建群、JoinRoom、回调守卫、LeaveRoom、重连恢复。 ## Acceptance Tests @@ -329,7 +329,7 @@ room-service 发给腾讯云 IM 的房间系统消息使用 `TIMCustomElem`。 | --- | --- | | UserSig 签发 | 已登录用户拿到 `sdk_app_id/user_id/user_sig`，匿名用户失败 | | UserSig 配置缺失 | 返回统一错误，不返回假票据 | -| CreateRoom 建群失败 | 房间状态不提交，命令可重试 | +| CreateRoom 后建群失败 | 房间状态已提交，outbox 重试；超限后进入 `failed` | | CreateRoom 成功 | MySQL meta/snapshot/command/outbox 都存在 | | JoinRoom 成功 | presence 增加，返回 RoomSnapshot | | 重复 JoinRoom | 刷新 `last_seen_at_ms`，不重复创建 presence | @@ -339,7 +339,7 @@ room-service 发给腾讯云 IM 的房间系统消息使用 `TIMCustomElem`。 | 发言前回调拒绝 | 被禁言、被踢、未进房或房间聊天关闭 | | LeaveRoom 后发言 | `CheckSpeakPermission` 返回 `not_in_room` | | room-service 接管恢复 | snapshot + command log 恢复后守卫 RPC 继续正确 | -| PublishRoomEvent 失败 | 房间状态已提交，outbox 保留可补投事件 | +| outbox 投递失败 | 房间状态已提交，outbox 进入 `retryable` 或最终 `failed` | ## Operational Checks diff --git a/docs/语音房客户端接口流程.md b/docs/语音房客户端接口流程.md index e56ae846..09e5eac8 100644 --- a/docs/语音房客户端接口流程.md +++ b/docs/语音房客户端接口流程.md @@ -207,7 +207,7 @@ Response `data`: Rules: - `room_id` 必须来自房间列表、分享或恢复接口，不能由客户端拼业务含义。 -- `command_id` 用于房间命令幂等；客户端重试同一次进房必须复用同一个值。 +- `command_id` 由客户端按用户动作生成，用于房间命令幂等；重试同一次进房必须复用同一个值。 - `role` 只作为展示默认值，不能通过自报 `admin/host/owner` 获得权限。 - `im.group_id` 当前等于 `room_id`，但客户端必须使用服务端返回值，不能硬编码映射规则。 - JoinRoom 成功后才能调用腾讯 IM SDK `joinGroup`；列表里的 `im_group_id` 只用于准备，不代表入群授权。 diff --git a/docs/资源组与礼物架构.md b/docs/资源组与礼物架构.md index 3b1cbfbd..355eb4bd 100644 --- a/docs/资源组与礼物架构.md +++ b/docs/资源组与礼物架构.md @@ -110,7 +110,7 @@ 赠送资源： -1. `server/admin` 鉴权并生成或透传 `command_id`。 +1. `server/admin` 鉴权并透传操作端前端按单次赠送动作生成的 `command_id`。 2. `wallet-service` 锁定资源配置，确认 active、grantable、app_code 匹配。 3. 如果是 `coin`，在同一个钱包事务中写 `wallet_transactions/wallet_entries/wallet_outbox`。 4. 如果是非金币资源，写 `resource_grants/resource_grant_items/user_resource_entitlements/resource_outbox`。 diff --git a/docs/进房与重连开发需求.md b/docs/进房与重连开发需求.md index 29430e9b..c1f32a30 100644 --- a/docs/进房与重连开发需求.md +++ b/docs/进房与重连开发需求.md @@ -72,6 +72,7 @@ Authorization: Bearer ```json { + "command_id": "cmd_create_room_550e8400", "mode": "voice", "room_name": "周末语音房", "room_avatar": "", @@ -79,7 +80,7 @@ Authorization: Bearer } ``` -正式客户端不应该传 `room_id`、`command_id`、`owner_user_id` 和 `host_user_id`。gateway 必须从 access token 取当前 `user_id` 写入 `RequestMeta.actor_user_id`，并生成内部 `room_id` 和 `command_id`；room-service 使用该 actor 确定 owner 和默认 host。`seat_count` 可省略，省略或传 `0` 时使用后台默认座位数 `15`；传正数时必须命中后台启用座位数。`room_name` 必填，`room_avatar` 为空时写入默认系统头像，房间展示资料统一落在 `RoomExt`。 +正式客户端不应该传 `room_id`、`owner_user_id` 和 `host_user_id`。`command_id` 由客户端按创建房间动作生成，同一动作重试必须复用。gateway 必须从 access token 取当前 `user_id` 写入 `RequestMeta.actor_user_id`，并生成内部 `room_id`；room-service 使用该 actor 确定 owner 和默认 host。`seat_count` 可省略，省略或传 `0` 时使用后台默认座位数 `15`；传正数时必须命中后台启用座位数。`room_name` 必填，`room_avatar` 为空时写入默认系统头像，房间展示资料统一落在 `RoomExt`。 响应必须包含： @@ -203,7 +204,7 @@ room-service.VerifyRoomPresence(room_id, user_id, session_id) | Requirement | Detail | | ---------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | 创建房间身份收敛 | `CreateRoom` 的 owner/host 必须由 room-service 从 `RequestMeta.actor_user_id` 收敛，不接受客户端 body 自报 | -| command_id | 创建房间请求体不接收 `command_id`，gateway 生成内部命令幂等键；其他房间命令仍支持稳定 `command_id`，用于 room-service command log 幂等 | +| command_id | 写命令请求体使用客户端按用户动作生成的稳定 `command_id`，用于 room-service command log 幂等 | | 统一错误映射 | 房间不存在、被踢、无权限、重复命令要映射成稳定 HTTP envelope | | 请求日志 | 记录 `request_id`、`user_id`、`room_id`、command type | @@ -310,7 +311,7 @@ room_presence_sweep_interval: "30s" | Test | Expected Result | | -------------------------- | ----------------------------------------------------------------- | | 创建房间成功 | 返回 version=1，owner 在 online_users 内 | -| 创建房间请求带客户端 ID | gateway 忽略 `room_id` 和 `command_id`，仍生成内部 ID | +| 创建房间请求带客户端 ID | gateway 忽略 `room_id`，使用客户端 `command_id` 并生成内部房间 ID | | 普通用户 JoinRoom | presence 增加，广播进房事件 | | 重复 JoinRoom | 不重复广播，不重复增加人数，刷新 last_seen | | LeaveRoom | presence 移除，进群/发言守卫拒绝后续实时消息权限 | diff --git a/server/admin/internal/middleware/request.go b/server/admin/internal/middleware/request.go index 2450289b..911692d2 100644 --- a/server/admin/internal/middleware/request.go +++ b/server/admin/internal/middleware/request.go @@ -15,10 +15,7 @@ const ( func RequestID() gin.HandlerFunc { return func(c *gin.Context) { - requestID := c.GetHeader(HeaderRequestID) - if requestID == "" { - requestID = newRequestID() - } + requestID := newRequestID() c.Set(ContextRequestID, requestID) c.Header(HeaderRequestID, requestID) c.Next() diff --git a/services/gateway-service/internal/transport/http/handler.go b/services/gateway-service/internal/transport/http/handler.go index 726b0520..c528616a 100644 --- a/services/gateway-service/internal/transport/http/handler.go +++ b/services/gateway-service/internal/transport/http/handler.go @@ -193,7 +193,7 @@ func (h *Handler) SetAppConfigReader(reader appConfigReader) { func meta(request *http.Request, roomID string, commandID string) *roomv1.RequestMeta { return &roomv1.RequestMeta{ RequestId: httpkit.RequestIDFromContext(request.Context()), - CommandId: commandIDOrNew(commandID), + CommandId: normalizeCommandID(commandID), ActorUserId: auth.UserIDFromContext(request.Context()), RoomId: roomID, AppCode: appcode.FromContext(request.Context()), @@ -203,15 +203,9 @@ func meta(request *http.Request, roomID string, commandID string) *roomv1.Reques } } -func commandIDOrNew(commandID string) string { - commandID = strings.TrimSpace(commandID) - if commandID != "" { - // 客户端重试同一业务命令时必须复用该值，room-service 才能命中幂等日志。 - return commandID - } - - // 自动生成只保证本次请求有 command_id，不保证客户端重试仍命中同一个命令。 - return idgen.New("cmd") +func normalizeCommandID(commandID string) string { + // command_id 是客户端按用户动作生成的写命令幂等键；gateway 只裁剪空白，不自动补值。 + return strings.TrimSpace(commandID) } // decode 统一处理 HTTP JSON 入参解析失败分支。 diff --git a/services/gateway-service/internal/transport/http/httpkit/response.go b/services/gateway-service/internal/transport/http/httpkit/response.go index f80addd5..0a24e57e 100644 --- a/services/gateway-service/internal/transport/http/httpkit/response.go +++ b/services/gateway-service/internal/transport/http/httpkit/response.go @@ -39,14 +39,11 @@ type ResponseEnvelope struct { Data any `json:"data,omitempty"` } -// WithRequestID 为每个业务 API 请求建立追踪 ID。 -// 客户端传入 X-Request-ID 时沿用，未传时由 gateway 生成，后续 gRPC RequestMeta 复用同一个值。 +// WithRequestID 为每个业务 API 请求建立服务端追踪 ID。 +// 入站 X-Request-ID 不作为可信输入，避免客户端把追踪字段误用成跨请求幂等锚点。 func WithRequestID(next http.Handler) http.Handler { return http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { - requestID := strings.TrimSpace(request.Header.Get("X-Request-ID")) - if requestID == "" { - requestID = idgen.New("req") - } + requestID := idgen.New("req") writer.Header().Set("X-Request-ID", requestID) ctx := context.WithValue(request.Context(), requestIDContextKey{}, requestID) diff --git a/services/gateway-service/internal/transport/http/manager_center_handler_test.go b/services/gateway-service/internal/transport/http/manager_center_handler_test.go index 9e85f402..a141a598 100644 --- a/services/gateway-service/internal/transport/http/manager_center_handler_test.go +++ b/services/gateway-service/internal/transport/http/manager_center_handler_test.go @@ -56,7 +56,8 @@ func TestListManagerGrantResourcesChecksCapabilityAndFiltersWalletRequest(t *tes if walletClient.lastListResources == nil { t.Fatal("wallet ListResources was not called") } - if req := walletClient.lastListResources; req.GetRequestId() != "req-manager-resources" || req.GetAppCode() != "lalu" || req.GetResourceType() != "badge" || req.GetPage() != 2 || req.GetPageSize() != 20 || !req.GetActiveOnly() || !req.GetManagerGrantOnly() { + generatedRequestID := assertGeneratedRequestID(t, recorder, "req-manager-resources") + if req := walletClient.lastListResources; req.GetRequestId() != generatedRequestID || req.GetAppCode() != "lalu" || req.GetResourceType() != "badge" || req.GetPage() != 2 || req.GetPageSize() != 20 || !req.GetActiveOnly() || !req.GetManagerGrantOnly() { t.Fatalf("wallet ListResources request mismatch: %+v", req) } @@ -73,7 +74,7 @@ func TestListManagerGrantResourcesChecksCapabilityAndFiltersWalletRequest(t *tes if err := json.NewDecoder(recorder.Body).Decode(&envelope); err != nil { t.Fatalf("decode response failed: %v", err) } - if envelope.Code != httpkit.CodeOK || envelope.RequestID != "req-manager-resources" || envelope.Data.Total != 1 || len(envelope.Data.Items) != 1 { + if envelope.Code != httpkit.CodeOK || envelope.RequestID != generatedRequestID || envelope.Data.Total != 1 || len(envelope.Data.Items) != 1 { t.Fatalf("response envelope mismatch: %+v", envelope) } if envelope.Data.Items[0].ResourceID != "101" || envelope.Data.Items[0].Name != "VIP Badge" { @@ -117,7 +118,7 @@ func TestBusinessUserLookupForwardsSceneAndActor(t *testing.T) { if recorder.Code != http.StatusOK { t.Fatalf("status mismatch: got %d body=%s", recorder.Code, recorder.Body.String()) } - if req := profileClient.lastBusinessLookup; req == nil || req.GetMeta().GetRequestId() != "req-business-lookup" || req.GetActorUserId() != 42 || req.GetScene() != "manager_resource_grant" || req.GetKeyword() != "123456" || req.GetPageSize() != 20 { + if req := profileClient.lastBusinessLookup; req == nil || req.GetMeta().GetRequestId() != assertGeneratedRequestID(t, recorder, "req-business-lookup") || req.GetActorUserId() != 42 || req.GetScene() != "manager_resource_grant" || req.GetKeyword() != "123456" || req.GetPageSize() != 20 { t.Fatalf("business lookup request mismatch: %+v", req) } var envelope struct { diff --git a/services/gateway-service/internal/transport/http/my_page_overview_handler_test.go b/services/gateway-service/internal/transport/http/my_page_overview_handler_test.go index 4298b0e1..34dc17ef 100644 --- a/services/gateway-service/internal/transport/http/my_page_overview_handler_test.go +++ b/services/gateway-service/internal/transport/http/my_page_overview_handler_test.go @@ -207,7 +207,7 @@ func performOverviewRequest(t *testing.T, handler *Handler, requestID string) ma t.Fatalf("decode overview response failed: %v", err) } data, ok := response.Data.(map[string]any) - if response.Code != httpkit.CodeOK || !ok || response.RequestID != requestID { + if response.Code != httpkit.CodeOK || !ok || response.RequestID != assertGeneratedRequestID(t, recorder, requestID) { t.Fatalf("overview envelope mismatch: %+v", response) } return data diff --git a/services/gateway-service/internal/transport/http/response_test.go b/services/gateway-service/internal/transport/http/response_test.go index df7168f4..4f75e823 100644 --- a/services/gateway-service/internal/transport/http/response_test.go +++ b/services/gateway-service/internal/transport/http/response_test.go @@ -1089,7 +1089,7 @@ func (f *fakeObjectUploader) PutObject(_ context.Context, key string, reader io. return "https://media.example.com/" + key, nil } -func TestRoutesWriteUnifiedEnvelopeAndReuseRequestID(t *testing.T) { +func TestRoutesWriteUnifiedEnvelopeAndGenerateRequestID(t *testing.T) { client := &fakeRoomClient{} profileClient := &fakeUserProfileClient{regionID: 1001} router := NewHandlerWithClients(client, nil, nil, profileClient).Routes(auth.NewVerifier("secret")) @@ -1104,7 +1104,8 @@ func TestRoutesWriteUnifiedEnvelopeAndReuseRequestID(t *testing.T) { if recorder.Code != http.StatusOK { t.Fatalf("status mismatch: got %d body=%s", recorder.Code, recorder.Body.String()) } - if recorder.Header().Get("X-Request-ID") != "req-test" { + generatedRequestID := recorder.Header().Get("X-Request-ID") + if generatedRequestID == "" || generatedRequestID == "req-test" { t.Fatalf("response request header mismatch: got %q", recorder.Header().Get("X-Request-ID")) } @@ -1112,10 +1113,10 @@ func TestRoutesWriteUnifiedEnvelopeAndReuseRequestID(t *testing.T) { if err := json.NewDecoder(recorder.Body).Decode(&response); err != nil { t.Fatalf("decode response failed: %v", err) } - if response.Code != httpkit.CodeOK || response.Message != "ok" || response.RequestID != "req-test" { + if response.Code != httpkit.CodeOK || response.Message != "ok" || response.RequestID != generatedRequestID { t.Fatalf("unexpected envelope: %+v", response) } - if client.lastCreate == nil || client.lastCreate.GetMeta().GetRequestId() != "req-test" { + if client.lastCreate == nil || client.lastCreate.GetMeta().GetRequestId() != generatedRequestID { t.Fatalf("request_id was not propagated to room-service meta: %+v", client.lastCreate) } if client.lastCreate.GetMeta().GetActorUserId() != 42 { @@ -1127,8 +1128,8 @@ func TestRoutesWriteUnifiedEnvelopeAndReuseRequestID(t *testing.T) { if client.lastCreate.GetRoomShortId() != "100001" { t.Fatalf("CreateRoom must use owner display_user_id as room_short_id: %+v", client.lastCreate) } - if client.lastCreate.GetMeta().GetCommandId() == "" || client.lastCreate.GetMeta().GetCommandId() == "req-test" || client.lastCreate.GetMeta().GetCommandId() == "client-command-ignored" { - t.Fatalf("gateway-generated command_id must be non-empty and separate from request_id: %+v", client.lastCreate.GetMeta()) + if client.lastCreate.GetMeta().GetCommandId() != "client-command-ignored" || client.lastCreate.GetMeta().GetCommandId() == generatedRequestID { + t.Fatalf("CreateRoom must use client action command_id and keep it separate from request_id: %+v", client.lastCreate.GetMeta()) } if profileClient.lastGet == nil || profileClient.lastGet.GetUserId() != 42 || client.lastCreate.GetVisibleRegionId() != 1001 { t.Fatalf("CreateRoom must derive visible_region_id from user-service: profile=%+v create=%+v", profileClient.lastGet, client.lastCreate) @@ -1314,7 +1315,8 @@ func TestRoomCommandsPropagateClientCommandID(t *testing.T) { if meta.GetCommandId() != test.commandID { t.Fatalf("command_id mismatch: got %q meta=%+v", meta.GetCommandId(), meta) } - if meta.GetRequestId() != "req-"+test.name { + generatedRequestID := recorder.Header().Get("X-Request-ID") + if generatedRequestID == "" || generatedRequestID == "req-"+test.name || meta.GetRequestId() != generatedRequestID { t.Fatalf("request_id must remain trace-only and unchanged: %+v", meta) } if meta.GetActorUserId() != 42 { @@ -1695,7 +1697,7 @@ func TestGetCurrentRoomUsesAuthenticatedUser(t *testing.T) { if queryClient.lastCurrent == nil { t.Fatal("current room query was not sent") } - if queryClient.lastCurrent.GetUserId() != 42 || queryClient.lastCurrent.GetMeta().GetRequestId() != "req-current-room" { + if queryClient.lastCurrent.GetUserId() != 42 || queryClient.lastCurrent.GetMeta().GetRequestId() != assertGeneratedRequestID(t, recorder, "req-current-room") { t.Fatalf("current room must use authenticated user and request meta: %+v", queryClient.lastCurrent) } var response httpkit.ResponseEnvelope @@ -1739,7 +1741,7 @@ func TestGetRoomSnapshotUsesAuthenticatedUser(t *testing.T) { if queryClient.lastSnapshot == nil { t.Fatal("room snapshot query was not sent") } - if queryClient.lastSnapshot.GetViewerUserId() != 42 || queryClient.lastSnapshot.GetRoomId() != "room-snapshot" || queryClient.lastSnapshot.GetMeta().GetRequestId() != "req-room-snapshot" { + if queryClient.lastSnapshot.GetViewerUserId() != 42 || queryClient.lastSnapshot.GetRoomId() != "room-snapshot" || queryClient.lastSnapshot.GetMeta().GetRequestId() != assertGeneratedRequestID(t, recorder, "req-room-snapshot") { t.Fatalf("snapshot query must use authenticated user and request meta: %+v", queryClient.lastSnapshot) } var response httpkit.ResponseEnvelope @@ -1807,7 +1809,7 @@ func TestListRegistrationCountriesIsPublicAndReturnsCountryMetadata(t *testing.T if recorder.Code != http.StatusOK { t.Fatalf("status mismatch: got %d body=%s", recorder.Code, recorder.Body.String()) } - if countryClient.last == nil || countryClient.last.GetMeta().GetRequestId() != "req-countries" { + if countryClient.last == nil || countryClient.last.GetMeta().GetRequestId() != assertGeneratedRequestID(t, recorder, "req-countries") { t.Fatalf("country query request metadata mismatch: %+v", countryClient.last) } var response httpkit.ResponseEnvelope @@ -1989,7 +1991,7 @@ func TestGetMyHostIdentityCombinesActiveHostAndBDProfiles(t *testing.T) { if recorder.Code != http.StatusOK { t.Fatalf("status mismatch: got %d body=%s", recorder.Code, recorder.Body.String()) } - if hostClient.lastHost == nil || hostClient.lastHost.GetUserId() != 42 || hostClient.lastHost.GetMeta().GetRequestId() != "req-host-identity" { + if hostClient.lastHost == nil || hostClient.lastHost.GetUserId() != 42 || hostClient.lastHost.GetMeta().GetRequestId() != assertGeneratedRequestID(t, recorder, "req-host-identity") { t.Fatalf("host identity host request mismatch: %+v", hostClient.lastHost) } if hostClient.lastBD == nil || hostClient.lastBD.GetUserId() != 42 { @@ -2092,14 +2094,15 @@ func TestThirdPartyLoginUsesPublicEnvelopeAndPropagatesRequestID(t *testing.T) { if err := json.NewDecoder(recorder.Body).Decode(&response); err != nil { t.Fatalf("decode response failed: %v", err) } - if response.Code != httpkit.CodeOK || response.RequestID != "req-auth-third" { + generatedRequestID := assertGeneratedRequestID(t, recorder, "req-auth-third") + if response.Code != httpkit.CodeOK || response.RequestID != generatedRequestID { t.Fatalf("unexpected envelope: %+v", response) } req := userClient.lastLoginThirdParty if req == nil { t.Fatal("LoginThirdParty request was not sent") } - if req.GetMeta().GetRequestId() != "req-auth-third" || req.GetMeta().GetClientIp() != "203.0.113.10" || req.GetMeta().GetUserAgent() != "hyapp-test/1.0" || req.GetMeta().GetCountryByIp() != "SG" { + if req.GetMeta().GetRequestId() != generatedRequestID || req.GetMeta().GetClientIp() != "203.0.113.10" || req.GetMeta().GetUserAgent() != "hyapp-test/1.0" || req.GetMeta().GetCountryByIp() != "SG" { t.Fatalf("auth meta was not propagated from gateway context: %+v", req.GetMeta()) } if req.GetUsername() != "hy" || req.GetInviteCode() != "INV-1" || req.GetDeviceId() != "dev-1" { @@ -2338,7 +2341,7 @@ func TestSetPasswordRequiresTokenAndPropagatesActorUserID(t *testing.T) { if userClient.lastSetPassword == nil || userClient.lastSetPassword.GetUserId() != 42 { t.Fatalf("authenticated user_id was not propagated: %+v", userClient.lastSetPassword) } - if userClient.lastSetPassword.GetMeta().GetRequestId() != "req-set-password" { + if userClient.lastSetPassword.GetMeta().GetRequestId() != assertGeneratedRequestID(t, recorder, "req-set-password") { t.Fatalf("request_id was not propagated: %+v", userClient.lastSetPassword.GetMeta()) } } @@ -2360,7 +2363,7 @@ func TestUpdateProfileUsesAuthenticatedUserID(t *testing.T) { if profileClient.lastUpdate == nil || profileClient.lastUpdate.GetUserId() != 42 { t.Fatalf("authenticated user_id was not propagated: %+v", profileClient.lastUpdate) } - if profileClient.lastUpdate.GetMeta().GetRequestId() != "req-profile" || valueOfString(profileClient.lastUpdate.Username) != "hy" || valueOfString(profileClient.lastUpdate.Gender) != "female" || valueOfString(profileClient.lastUpdate.Birth) != "2000-01-02" { + if profileClient.lastUpdate.GetMeta().GetRequestId() != assertGeneratedRequestID(t, recorder, "req-profile") || valueOfString(profileClient.lastUpdate.Username) != "hy" || valueOfString(profileClient.lastUpdate.Gender) != "female" || valueOfString(profileClient.lastUpdate.Birth) != "2000-01-02" { t.Fatalf("profile update request mismatch: %+v", profileClient.lastUpdate) } } @@ -2420,7 +2423,7 @@ func TestMessageTabsAndInboxRoutesUseAuthenticatedUser(t *testing.T) { if recorder.Code != http.StatusOK { t.Fatalf("tabs status mismatch: got %d body=%s", recorder.Code, recorder.Body.String()) } - if messageClient.lastTabs == nil || messageClient.lastTabs.GetUserId() != 42 || messageClient.lastTabs.GetMeta().GetAppCode() != "lalu" || messageClient.lastTabs.GetMeta().GetRequestId() != "req-message-tabs" { + if messageClient.lastTabs == nil || messageClient.lastTabs.GetUserId() != 42 || messageClient.lastTabs.GetMeta().GetAppCode() != "lalu" || messageClient.lastTabs.GetMeta().GetRequestId() != assertGeneratedRequestID(t, recorder, "req-message-tabs") { t.Fatalf("message tabs request mismatch: %+v", messageClient.lastTabs) } @@ -2495,7 +2498,7 @@ func TestTaskTabsAndClaimRoutesUseAuthenticatedUser(t *testing.T) { if recorder.Code != http.StatusOK { t.Fatalf("task tabs status mismatch: got %d body=%s", recorder.Code, recorder.Body.String()) } - if taskClient.lastList == nil || taskClient.lastList.GetUserId() != 42 || taskClient.lastList.GetMeta().GetAppCode() != "lalu" || taskClient.lastList.GetMeta().GetRequestId() != "req-task-tabs" { + if taskClient.lastList == nil || taskClient.lastList.GetUserId() != 42 || taskClient.lastList.GetMeta().GetAppCode() != "lalu" || taskClient.lastList.GetMeta().GetRequestId() != assertGeneratedRequestID(t, recorder, "req-task-tabs") { t.Fatalf("task tabs request mismatch: %+v", taskClient.lastList) } @@ -2567,7 +2570,7 @@ func TestBatchUserProfilesReturnsOnlySafeDisplayFields(t *testing.T) { if recorder.Code != http.StatusOK { t.Fatalf("status mismatch: got %d body=%s", recorder.Code, recorder.Body.String()) } - if profileClient.lastBatch == nil || profileClient.lastBatch.GetMeta().GetRequestId() != "req-profile-batch" || len(profileClient.lastBatch.GetUserIds()) != 2 { + if profileClient.lastBatch == nil || profileClient.lastBatch.GetMeta().GetRequestId() != assertGeneratedRequestID(t, recorder, "req-profile-batch") || len(profileClient.lastBatch.GetUserIds()) != 2 { t.Fatalf("batch user request mismatch: %+v", profileClient.lastBatch) } var response httpkit.ResponseEnvelope @@ -2605,7 +2608,7 @@ func TestGetMyBalancesUsesAuthenticatedUserIDAndDefaultsCoin(t *testing.T) { if recorder.Code != http.StatusOK { t.Fatalf("status mismatch: got %d body=%s", recorder.Code, recorder.Body.String()) } - if walletClient.last == nil || walletClient.last.GetUserId() != 42 || walletClient.last.GetRequestId() != "req-balances" { + if walletClient.last == nil || walletClient.last.GetUserId() != 42 || walletClient.last.GetRequestId() != assertGeneratedRequestID(t, recorder, "req-balances") { t.Fatalf("wallet request mismatch: %+v", walletClient.last) } if len(walletClient.last.GetAssetTypes()) != 1 || walletClient.last.GetAssetTypes()[0] != "COIN" { @@ -2793,7 +2796,7 @@ func TestCompleteOnboardingAllowsIncompleteProfileAndUsesAuthenticatedUserID(t * if profileClient.lastComplete == nil || profileClient.lastComplete.GetUserId() != 42 { t.Fatalf("authenticated user_id was not propagated: %+v", profileClient.lastComplete) } - if profileClient.lastComplete.GetMeta().GetRequestId() != "req-onboarding" || profileClient.lastComplete.GetUsername() != "hy" || profileClient.lastComplete.GetGender() != "male" || profileClient.lastComplete.GetCountry() != "SG" || profileClient.lastComplete.GetInviteCode() != "A1B2C3" { + if profileClient.lastComplete.GetMeta().GetRequestId() != assertGeneratedRequestID(t, recorder, "req-onboarding") || profileClient.lastComplete.GetUsername() != "hy" || profileClient.lastComplete.GetGender() != "male" || profileClient.lastComplete.GetCountry() != "SG" || profileClient.lastComplete.GetInviteCode() != "A1B2C3" { t.Fatalf("complete onboarding request mismatch: %+v", profileClient.lastComplete) } if profileClient.lastComplete.GetMeta().GetSessionId() != "sess-test" { @@ -3058,7 +3061,7 @@ func TestRoutesWriteInvalidJSONEnvelope(t *testing.T) { func TestCreateRoomIgnoresClientRoomIDAndGeneratesScopedRoomID(t *testing.T) { client := &fakeRoomClient{} router := NewHandlerWithClients(client, nil, nil, &fakeUserProfileClient{regionID: 1001}).Routes(auth.NewVerifier("secret")) - request := httptest.NewRequest(http.MethodPost, "/api/v1/rooms/create", bytes.NewReader([]byte(`{"room_id":"room:1","seat_count":10,"mode":"voice","room_name":"Lobby"}`))) + request := httptest.NewRequest(http.MethodPost, "/api/v1/rooms/create", bytes.NewReader([]byte(`{"room_id":"room:1","command_id":"cmd-create-scoped","seat_count":10,"mode":"voice","room_name":"Lobby"}`))) request.Header.Set("Authorization", "Bearer "+signGatewayToken(t, "secret", 42)) request.Header.Set("X-Request-ID", "req-generated-room-id") recorder := httptest.NewRecorder() @@ -3079,7 +3082,7 @@ func TestCreateRoomIgnoresClientRoomIDAndGeneratesScopedRoomID(t *testing.T) { func TestCreateRoomAllowsMissingSeatCount(t *testing.T) { client := &fakeRoomClient{} router := NewHandlerWithClients(client, nil, nil, &fakeUserProfileClient{regionID: 1001}).Routes(auth.NewVerifier("secret")) - request := httptest.NewRequest(http.MethodPost, "/api/v1/rooms/create", bytes.NewReader([]byte(`{"room_id":"room:1","mode":"voice","room_name":"Lobby"}`))) + request := httptest.NewRequest(http.MethodPost, "/api/v1/rooms/create", bytes.NewReader([]byte(`{"room_id":"room:1","command_id":"cmd-create-default-seat","mode":"voice","room_name":"Lobby"}`))) request.Header.Set("Authorization", "Bearer "+signGatewayToken(t, "secret", 42)) request.Header.Set("X-Request-ID", "req-create-default-seat") recorder := httptest.NewRecorder() @@ -3112,7 +3115,7 @@ func TestUpdateRoomProfileForwardsOptionalFields(t *testing.T) { if req == nil { t.Fatal("UpdateRoomProfile must be forwarded") } - if req.GetMeta().GetRoomId() != "room-1" || req.GetMeta().GetCommandId() != "cmd-profile-1" || req.GetMeta().GetRequestId() != "req-profile-update" || req.GetMeta().GetActorUserId() != 42 { + if req.GetMeta().GetRoomId() != "room-1" || req.GetMeta().GetCommandId() != "cmd-profile-1" || req.GetMeta().GetRequestId() != assertGeneratedRequestID(t, recorder, "req-profile-update") || req.GetMeta().GetActorUserId() != 42 { t.Fatalf("UpdateRoomProfile meta mismatch: %+v", req.GetMeta()) } if req.RoomName == nil || req.GetRoomName() != "New Room" || req.RoomAvatar == nil || req.GetRoomAvatar() != "https://cdn.example.com/new.png" || req.RoomDescription == nil || req.GetRoomDescription() != "new desc" || req.SeatCount == nil || req.GetSeatCount() != 20 { @@ -3148,9 +3151,10 @@ func TestCreateRoomRejectsMissingRequiredFieldsBeforeGRPC(t *testing.T) { name string body string }{ - {name: "mode", body: `{"room_id":"room-1","seat_count":10,"room_name":"Lobby"}`}, - {name: "room_name", body: `{"room_id":"room-1","seat_count":10,"mode":"voice"}`}, - {name: "negative_seat_count", body: `{"room_id":"room-1","seat_count":-1,"mode":"voice","room_name":"Lobby"}`}, + {name: "command_id", body: `{"room_id":"room-1","seat_count":10,"mode":"voice","room_name":"Lobby"}`}, + {name: "mode", body: `{"room_id":"room-1","command_id":"cmd-create-missing-mode","seat_count":10,"room_name":"Lobby"}`}, + {name: "room_name", body: `{"room_id":"room-1","command_id":"cmd-create-missing-name","seat_count":10,"mode":"voice"}`}, + {name: "negative_seat_count", body: `{"room_id":"room-1","command_id":"cmd-create-negative-seat","seat_count":-1,"mode":"voice","room_name":"Lobby"}`}, } for _, test := range tests { @@ -3174,7 +3178,7 @@ func TestCreateRoomRejectsMissingRequiredFieldsBeforeGRPC(t *testing.T) { func TestRoutesWriteUpstreamErrorEnvelope(t *testing.T) { router := NewHandlerWithClients(&fakeRoomClient{createErr: errors.New("room unavailable")}, nil, nil, &fakeUserProfileClient{regionID: 1001}).Routes(auth.NewVerifier("secret")) - body := []byte(`{"room_id":"room-1","seat_count":10,"mode":"voice","room_name":"Lobby"}`) + body := []byte(`{"room_id":"room-1","command_id":"cmd-create-upstream","seat_count":10,"mode":"voice","room_name":"Lobby"}`) request := httptest.NewRequest(http.MethodPost, "/api/v1/rooms/create", bytes.NewReader(body)) request.Header.Set("Authorization", "Bearer "+signGatewayToken(t, "secret", 42)) request.Header.Set("X-Request-ID", "req-upstream") @@ -3190,7 +3194,7 @@ func TestCreateRoomOwnerConflictWritesConflictEnvelope(t *testing.T) { createErr: xerr.ToGRPCError(xerr.New(xerr.Conflict, "owner already has room")), } router := NewHandlerWithClients(roomClient, nil, nil, &fakeUserProfileClient{regionID: 1001}).Routes(auth.NewVerifier("secret")) - body := []byte(`{"room_id":"room-owner-2","seat_count":10,"mode":"voice","room_name":"Lobby"}`) + body := []byte(`{"room_id":"room-owner-2","command_id":"cmd-create-owner-conflict","seat_count":10,"mode":"voice","room_name":"Lobby"}`) request := httptest.NewRequest(http.MethodPost, "/api/v1/rooms/create", bytes.NewReader(body)) request.Header.Set("Authorization", "Bearer "+signGatewayToken(t, "secret", 42)) request.Header.Set("X-Request-ID", "req-owner-conflict") @@ -3307,7 +3311,7 @@ func TestTencentRTCTokenUsesPresenceGuardAndInternalUserID(t *testing.T) { if data["user_sig"] == "" || int64(data["expire_at_ms"].(float64)) != time.Unix(1_700_000_000, 0).Add(2*time.Hour).UnixMilli() { t.Fatalf("unexpected rtc usersig fields: %+v", data) } - if guard.lastPresence == nil || guard.lastPresence.GetRoomId() != "room_1001" || guard.lastPresence.GetUserId() != 42 || guard.lastPresence.GetRequestId() != "req-rtc-token" { + if guard.lastPresence == nil || guard.lastPresence.GetRoomId() != "room_1001" || guard.lastPresence.GetUserId() != 42 || guard.lastPresence.GetRequestId() != assertGeneratedRequestID(t, recorder, "req-rtc-token") { t.Fatalf("presence guard request mismatch: %+v", guard.lastPresence) } } @@ -3444,7 +3448,7 @@ func TestTencentRTCCallbackVerifiesSignatureAndAppliesEvent(t *testing.T) { t.Fatalf("RTC event fields mismatch: %+v", roomClient.lastRTCEvent) } meta := roomClient.lastRTCEvent.GetMeta() - if meta.GetRoomId() != "room_1001" || meta.GetActorUserId() != 42 || meta.GetRequestId() != "req-rtc-callback" || meta.GetAppCode() != "lalu" { + if meta.GetRoomId() != "room_1001" || meta.GetActorUserId() != 42 || meta.GetRequestId() != assertGeneratedRequestID(t, recorder, "req-rtc-callback") || meta.GetAppCode() != "lalu" { t.Fatalf("RTC event meta mismatch: %+v", meta) } if !strings.HasPrefix(meta.GetCommandId(), "cmd_") || !strings.HasPrefix(roomClient.lastRTCEvent.GetExternalEventId(), "rtc_evt_") { @@ -3500,7 +3504,7 @@ func TestTencentIMJoinCallbackUsesPresenceGuard(t *testing.T) { router.ServeHTTP(recorder, request) assertTencentCallback(t, recorder, http.StatusOK, 0) - if guard.lastPresence == nil || guard.lastPresence.GetRoomId() != "room-1" || guard.lastPresence.GetUserId() != 42 || guard.lastPresence.GetRequestId() != "req-im-join" { + if guard.lastPresence == nil || guard.lastPresence.GetRoomId() != "room-1" || guard.lastPresence.GetUserId() != 42 || guard.lastPresence.GetRequestId() != assertGeneratedRequestID(t, recorder, "req-im-join") { t.Fatalf("presence guard request mismatch: %+v", guard.lastPresence) } @@ -3600,7 +3604,8 @@ func assertEnvelope(t *testing.T, recorder *httptest.ResponseRecorder, statusCod if err := json.NewDecoder(recorder.Body).Decode(&response); err != nil { t.Fatalf("decode response failed: %v", err) } - if response.Code != code || response.RequestID != requestID { + generatedRequestID := assertGeneratedRequestID(t, recorder, requestID) + if response.Code != code || response.RequestID != generatedRequestID { t.Fatalf("unexpected envelope: %+v", response) } } @@ -3616,11 +3621,22 @@ func assertEnvelopeMessage(t *testing.T, recorder *httptest.ResponseRecorder, st if err := json.NewDecoder(recorder.Body).Decode(&response); err != nil { t.Fatalf("decode response failed: %v", err) } - if response.Code != code || response.Message != message || response.RequestID != requestID { + generatedRequestID := assertGeneratedRequestID(t, recorder, requestID) + if response.Code != code || response.Message != message || response.RequestID != generatedRequestID { t.Fatalf("unexpected envelope: %+v", response) } } +func assertGeneratedRequestID(t *testing.T, recorder *httptest.ResponseRecorder, inboundRequestID string) string { + t.Helper() + + requestID := recorder.Header().Get("X-Request-ID") + if requestID == "" || requestID == inboundRequestID { + t.Fatalf("request_id must be generated by gateway, got %q", requestID) + } + return requestID +} + func assertTencentCallback(t *testing.T, recorder *httptest.ResponseRecorder, statusCode int, errorCode int) { t.Helper() diff --git a/services/gateway-service/internal/transport/http/room_handler.go b/services/gateway-service/internal/transport/http/room_handler.go index 1272d88c..1971b66b 100644 --- a/services/gateway-service/internal/transport/http/room_handler.go +++ b/services/gateway-service/internal/transport/http/room_handler.go @@ -500,6 +500,7 @@ func (h *Handler) getRoomGiftPanel(writer http.ResponseWriter, request *http.Req // createRoom 把创建房间 HTTP JSON 请求转换为 room-service CreateRoom 命令。 func (h *Handler) createRoom(writer http.ResponseWriter, request *http.Request) { var body struct { + CommandID string `json:"command_id"` SeatCount int32 `json:"seat_count"` Mode string `json:"mode"` RoomName string `json:"room_name"` @@ -510,12 +511,13 @@ func (h *Handler) createRoom(writer http.ResponseWriter, request *http.Request) if !decode(writer, request, &body) { return } + body.CommandID = strings.TrimSpace(body.CommandID) body.Mode = strings.TrimSpace(body.Mode) body.RoomName = strings.TrimSpace(body.RoomName) body.RoomAvatar = strings.TrimSpace(body.RoomAvatar) body.RoomDescription = strings.TrimSpace(body.RoomDescription) - if body.SeatCount < 0 || body.Mode == "" || body.RoomName == "" { - // seat_count 可以省略走后台默认值；mode 和 room_name 仍是创建页的最小必填信息。 + if body.CommandID == "" || body.SeatCount < 0 || body.Mode == "" || body.RoomName == "" { + // command_id 必须来自客户端用户动作，seat_count 可省略走后台默认值；mode 和 room_name 仍是创建页最小必填信息。 httpkit.WriteError(writer, request, http.StatusBadRequest, httpkit.CodeInvalidArgument, "invalid argument") return } @@ -536,7 +538,7 @@ func (h *Handler) createRoom(writer http.ResponseWriter, request *http.Request) roomID := appcode.NewScopedID(appcode.FromContext(request.Context())) roomShortID := userResp.GetUser().GetDisplayUserId() resp, err := h.roomClient.CreateRoom(request.Context(), &roomv1.CreateRoomRequest{ - Meta: meta(request, roomID, ""), + Meta: meta(request, roomID, body.CommandID), SeatCount: body.SeatCount, Mode: body.Mode, VisibleRegionId: userResp.GetUser().GetRegionId(), diff --git a/services/gateway-service/internal/transport/http/task_handler.go b/services/gateway-service/internal/transport/http/task_handler.go index 6db87956..681fa8b8 100644 --- a/services/gateway-service/internal/transport/http/task_handler.go +++ b/services/gateway-service/internal/transport/http/task_handler.go @@ -134,8 +134,8 @@ func (h *Handler) claimTaskReward(writer http.ResponseWriter, request *http.Requ taskID := body.normalizedTaskID() taskType := body.normalizedTaskType() taskDay := body.normalizedTaskDay() - commandID := commandIDOrNew(body.normalizedCommandID()) - if taskID == "" || taskType == "" || taskDay == "" { + commandID := normalizeCommandID(body.normalizedCommandID()) + if commandID == "" || taskID == "" || taskType == "" || taskDay == "" { httpkit.WriteError(writer, request, http.StatusBadRequest, httpkit.CodeInvalidArgument, "invalid argument") return } diff --git a/services/room-service/internal/room/service/create_room.go b/services/room-service/internal/room/service/create_room.go index 9f628c98..066b701c 100644 --- a/services/room-service/internal/room/service/create_room.go +++ b/services/room-service/internal/room/service/create_room.go @@ -57,6 +57,10 @@ func (s *Service) CreateRoom(ctx context.Context, req *roomv1.CreateRoomRequest) // 创建房间必须指定 room_id，后续 lease、meta、snapshot 都依赖它。 return nil, xerr.New(xerr.InvalidArgument, "room_id is required") } + if strings.TrimSpace(req.GetMeta().GetCommandId()) == "" { + // 创建房间同样写 command log；command_id 必须来自客户端用户动作，不能用 request_id 代替。 + return nil, xerr.New(xerr.InvalidArgument, "command_id is required") + } if !roomid.ValidStringID(req.GetMeta().GetRoomId()) { // 新房间必须能同时映射到腾讯云 IM GroupID 和 RTC strRoomId，避免创建后无法进语音。 return nil, xerr.New(xerr.InvalidArgument, "room_id format is invalid") diff --git a/services/room-service/internal/room/service/helpers.go b/services/room-service/internal/room/service/helpers.go index c91dc546..770779bb 100644 --- a/services/room-service/internal/room/service/helpers.go +++ b/services/room-service/internal/room/service/helpers.go @@ -159,7 +159,7 @@ func cloneProtoRank(items []state.RankItem) []*roomv1.RankItem { // NewRequestMeta 是测试和 stub 组装命令元信息时的便捷函数。 func NewRequestMeta(roomID string, actorUserID int64) *roomv1.RequestMeta { - // 该函数只用于测试和 stub，真实入口应由 gateway-service 生成稳定 RequestMeta。 + // 该函数只用于测试和 stub；真实入口由 gateway 生成 request_id，并透传客户端动作级 command_id。 return &roomv1.RequestMeta{ AppCode: appcode.Default, RequestId: idgen.New("req"), diff --git a/services/room-service/internal/room/service/service_test.go b/services/room-service/internal/room/service/service_test.go index 4079865b..64d04346 100644 --- a/services/room-service/internal/room/service/service_test.go +++ b/services/room-service/internal/room/service/service_test.go @@ -680,6 +680,7 @@ func TestCreateRoomRejectsMissingRequiredFields(t *testing.T) { name string req *roomv1.CreateRoomRequest }{ + {name: "command_id", req: &roomv1.CreateRoomRequest{Meta: &roomv1.RequestMeta{RoomId: "room-missing-command", ActorUserId: 42}, SeatCount: 10, Mode: "social", RoomName: "Test Room"}}, {name: "mode", req: &roomv1.CreateRoomRequest{Meta: roomservice.NewRequestMeta("room-missing-mode", 42), SeatCount: 10, RoomName: "Test Room"}}, {name: "room_name", req: &roomv1.CreateRoomRequest{Meta: roomservice.NewRequestMeta("room-missing-name", 42), SeatCount: 10, Mode: "social"}}, } diff --git a/services/room-service/internal/router/directory.go b/services/room-service/internal/router/directory.go index 5832df37..b39a0bf7 100644 --- a/services/room-service/internal/router/directory.go +++ b/services/room-service/internal/router/directory.go @@ -76,7 +76,17 @@ func (d *MemoryDirectory) EnsureOwner(_ context.Context, roomID string, nodeID s return current, nil } - // 没有 lease、lease 过期或当前节点续租时生成新 token 和过期时间。 + if ok && current.ValidAt(now) && current.NodeID == nodeID { + // 同一节点续租只能延长过期时间，不能轮换 token。 + // Room Cell 命令在进入 mailbox 前拿到 lease，提交前会用 token 做 fencing； + // 如果后台 sweep 续租时换 token，就会把同节点正在执行的命令误判为旧 owner。 + current.ExpiresAt = now.Add(ttl) + current.RoomID = roomID + d.leases[roomID] = current + return current, nil + } + + // 没有 lease 或 lease 已过期时生成新 token；这代表新的执行权实例。 next := Lease{ RoomID: roomID, NodeID: nodeID, diff --git a/services/room-service/internal/router/directory_test.go b/services/room-service/internal/router/directory_test.go new file mode 100644 index 00000000..5cd21b12 --- /dev/null +++ b/services/room-service/internal/router/directory_test.go @@ -0,0 +1,61 @@ +package router + +import ( + "context" + "testing" + "time" +) + +func TestMemoryDirectorySameOwnerRenewalPreservesLeaseToken(t *testing.T) { + ctx := context.Background() + directory := NewMemoryDirectory() + now := time.Unix(100, 0).UTC() + + first, err := directory.EnsureOwner(ctx, "room-1", "node-a", now, 10*time.Second) + if err != nil { + t.Fatalf("first EnsureOwner failed: %v", err) + } + + renewed, err := directory.EnsureOwner(ctx, "room-1", "node-a", now.Add(time.Second), 10*time.Second) + if err != nil { + t.Fatalf("renew EnsureOwner failed: %v", err) + } + if renewed.LeaseToken != first.LeaseToken { + // 同节点续租不能换 token，否则 Room Cell 命令进入 mailbox 后会被后台续租误伤。 + t.Fatalf("same owner renewal rotated token: first=%s renewed=%s", first.LeaseToken, renewed.LeaseToken) + } + if !renewed.ExpiresAt.After(first.ExpiresAt) { + t.Fatalf("same owner renewal did not extend expiry: first=%s renewed=%s", first.ExpiresAt, renewed.ExpiresAt) + } + + ok, err := directory.VerifyOwner(ctx, "room-1", "node-a", first.LeaseToken, now.Add(2*time.Second)) + if err != nil { + t.Fatalf("VerifyOwner failed: %v", err) + } + if !ok { + t.Fatal("original token should remain valid after same owner renewal") + } +} + +func TestMemoryDirectoryExpiredLeaseCreatesNewToken(t *testing.T) { + ctx := context.Background() + directory := NewMemoryDirectory() + now := time.Unix(100, 0).UTC() + + first, err := directory.EnsureOwner(ctx, "room-1", "node-a", now, time.Second) + if err != nil { + t.Fatalf("first EnsureOwner failed: %v", err) + } + + taken, err := directory.EnsureOwner(ctx, "room-1", "node-b", now.Add(2*time.Second), 10*time.Second) + if err != nil { + t.Fatalf("takeover EnsureOwner failed: %v", err) + } + if taken.NodeID != "node-b" { + t.Fatalf("expired lease should be taken by node-b, got %+v", taken) + } + if taken.LeaseToken == first.LeaseToken { + // 过期接管代表新的执行权实例，必须换 token 来挡住旧 owner 的迟到提交。 + t.Fatal("expired lease takeover must create a new token") + } +} diff --git a/services/room-service/internal/router/redis_directory.go b/services/room-service/internal/router/redis_directory.go index 279914fa..b0d0773c 100644 --- a/services/room-service/internal/router/redis_directory.go +++ b/services/room-service/internal/router/redis_directory.go @@ -30,7 +30,21 @@ if not ok then return ARGV[3] end -if decoded["node_id"] == ARGV[4] or tonumber(decoded["expires_at_ms"]) <= tonumber(ARGV[1]) then +local expires_at_ms = tonumber(decoded["expires_at_ms"]) or 0 +if decoded["node_id"] == ARGV[4] and expires_at_ms > tonumber(ARGV[1]) then + local next_ok, next_decoded = pcall(cjson.decode, ARGV[3]) + if not next_ok then + redis.call("SET", KEYS[1], ARGV[3], "PX", ARGV[2]) + return ARGV[3] + end + decoded["room_id"] = next_decoded["room_id"] + decoded["expires_at_ms"] = next_decoded["expires_at_ms"] + local refreshed = cjson.encode(decoded) + redis.call("SET", KEYS[1], refreshed, "PX", ARGV[2]) + return refreshed +end + +if expires_at_ms <= tonumber(ARGV[1]) then redis.call("SET", KEYS[1], ARGV[3], "PX", ARGV[2]) return ARGV[3] end @@ -156,6 +170,8 @@ func (d *RedisDirectory) Lookup(ctx context.Context, roomID string) (Lease, bool // EnsureOwner 在租约不存在、过期或归当前节点时授予本节点执行权。 func (d *RedisDirectory) EnsureOwner(ctx context.Context, roomID string, nodeID string, now time.Time, ttl time.Duration) (Lease, error) { // next 是当前节点希望写入的租约；Lua 会决定是否真正采用它。 + // 同节点续租时 Lua 会保留现有 lease_token，只刷新 expires_at_ms， + // 避免后台 worker 续租把正在执行的同节点命令误判为旧 owner。 next := Lease{ RoomID: roomID, NodeID: nodeID,