新增部署相关

This commit is contained in:
zhx 2026-05-12 21:51:39 +08:00
parent 8955a0d118
commit 1e542ba563
83 changed files with 2344 additions and 3719 deletions

View File

@ -143,15 +143,18 @@ MySQL 是本地和线上运行的持久化底座:
`room-service` uses Redis for current room ownership:
```text
room:route:{room_id}
room:route:{app_code + "\0" + room_id}
room:node:{node_id}
```
Redis lease 只保存当前执行权,不能作为恢复来源。恢复来源是 MySQL snapshot + command log。
Redis lease 只保存当前执行权和节点转发地址,不能作为恢复来源。恢复来源是 MySQL snapshot + command log。
## Room Recovery
A room is not permanently bound to one process. A node is only the current executor while its Redis lease is valid.
在多机部署下gateway 可以把 room gRPC 请求打到 `room-service` 内网 CLB。非 owner 节点会读取 `room:route:*``room:node:*`把房间命令、IM guard 和完整快照查询转发到当前 owner 实例。每个 room 节点必须配置唯一 `node_id` 和实例级 `advertise_addr``advertise_addr` 不能写负载均衡地址。无 Kubernetes 的部署步骤见 `docs/多机部署与无感发布.md`
Recovery path:
1. A request reaches a node that can own the room.

View File

@ -0,0 +1,71 @@
# Standalone CVM Deployment Templates
本目录提供不依赖 Kubernetes 的生产部署模板。目标形态是 systemd 托管 Docker 容器,公网 CLB 只接 gateway内部服务通过内网 CLB/DNS 或固定私网地址访问。
## Layout
- `systemd/*.service`: 每个后端服务一份 systemd unit负责启动、停止和故障重启 Docker 容器。
- `docker/*.env.example`: 每个服务一份 Docker 运行参数示例,只保存镜像名、容器名、配置路径和停止等待时间。
- 服务业务配置仍使用 `services/<service>/configs/config.tencent.example.yaml` 作为线上模板,渲染后放到 `/etc/hyapp/<service>/config.yaml`
## Install One Service
`gateway-service` 为例:
```bash
sudo mkdir -p /etc/hyapp/gateway-service
sudo cp deploy/standalone/docker/gateway-service.env.example /etc/hyapp/gateway-service/docker.env
sudo cp services/gateway-service/configs/config.tencent.example.yaml /etc/hyapp/gateway-service/config.yaml
sudo vi /etc/hyapp/gateway-service/docker.env
sudo vi /etc/hyapp/gateway-service/config.yaml
sudo cp deploy/standalone/systemd/hyapp-gateway-service.service /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable --now hyapp-gateway-service
sudo systemctl status hyapp-gateway-service
sudo docker inspect --format '{{json .State.Health}}' hyapp-gateway-service
```
`docker.env` 不承载业务密钥;密钥仍在最终 `config.yaml` 中,或者由发布系统先从密钥系统渲染到 `config.yaml`。当前代码只读取单个 YAML不会自动展开 `${VAR}`
## Network
模板默认使用 `--network=host`,让容器直接监听项目约定的 `13xxx` 端口。这样 CLB、安全组和本机排障都以同一个端口工作
- `13000`: gateway HTTP
- `13001`: room gRPC
- `13004`: wallet gRPC
- `13005`: user gRPC
- `13006`: activity gRPC
- `13007`: cron gRPC
- `13008`: game gRPC
同一台 CVM 上不要启动两个相同端口的同服务实例。做蓝绿发布时优先使用新 CVM 或新容器节点,健康通过后再挂载到 CLB。
## Rolling Update
普通服务的无感更新流程:
1. 在新 CVM 上准备 `/etc/hyapp/<service>/docker.env``/etc/hyapp/<service>/config.yaml`
2. `systemctl start hyapp-<service>`,等待 Docker health 进入 `healthy`
3. 将新实例加入对应 CLB 或内部服务入口。
4. 观察错误率、延迟和服务日志。
5. 从 CLB 摘除旧实例,等待连接保护或业务 drain 窗口结束。
6. `systemctl stop hyapp-<service>`Docker 会先发送 SIGTERM服务内的 graceful shutdown 会进入 draining 并等待已进入请求结束。
`room-service` 当前还没有 gateway owner-routing 或非 owner 自动转发能力。没有补完该能力前,不要把多个 `room-service` active 实例直接放到普通负载均衡后面承接同一批房间命令;生产先保持单 active或者只把备用实例作为故障接管目标。
## Service Discovery
不使用 Kubernetes 时,第一阶段推荐用内网 CLB/DNS 提供稳定入口,例如:
```yaml
user_service_addr: "user.service.internal:13005"
wallet_service_addr: "wallet.service.internal:13004"
activity_service_addr: "activity.service.internal:13006"
room_service_addr: "room.service.internal:13001"
game_service_addr: "game.service.internal:13008"
```
如果后续引入 Consul/Nacos仍建议让业务配置只保存稳定服务名不把实例 IP 写进每个服务配置。

View File

@ -0,0 +1,6 @@
# Image is immutable per release; publish scripts pull this tag before restarting systemd.
IMAGE=ccr.ccs.tencentyun.com/hyapp/activity-service:REPLACE_WITH_RELEASE_TAG
CONTAINER_NAME=hyapp-activity-service
CONFIG_PATH=/etc/hyapp/activity-service/config.yaml
STOP_TIMEOUT_SEC=60

View File

@ -0,0 +1,6 @@
# cron-service already uses MySQL task leases, but keep one replica until every task is verified under multi-node scheduling.
IMAGE=ccr.ccs.tencentyun.com/hyapp/cron-service:REPLACE_WITH_RELEASE_TAG
CONTAINER_NAME=hyapp-cron-service
CONFIG_PATH=/etc/hyapp/cron-service/config.yaml
STOP_TIMEOUT_SEC=90

View File

@ -0,0 +1,5 @@
IMAGE=ccr.ccs.tencentyun.com/hyapp/game-service:REPLACE_WITH_RELEASE_TAG
CONTAINER_NAME=hyapp-game-service
CONFIG_PATH=/etc/hyapp/game-service/config.yaml
STOP_TIMEOUT_SEC=60

View File

@ -0,0 +1,6 @@
# gateway instances are stateless HTTP entries and should be attached behind the public CLB only after /healthz/ready passes.
IMAGE=ccr.ccs.tencentyun.com/hyapp/gateway-service:REPLACE_WITH_RELEASE_TAG
CONTAINER_NAME=hyapp-gateway-service
CONFIG_PATH=/etc/hyapp/gateway-service/config.yaml
STOP_TIMEOUT_SEC=60

View File

@ -0,0 +1,6 @@
# Keep room-service single-active until owner routing or non-owner forwarding is implemented.
IMAGE=ccr.ccs.tencentyun.com/hyapp/room-service:REPLACE_WITH_RELEASE_TAG
CONTAINER_NAME=hyapp-room-service
CONFIG_PATH=/etc/hyapp/room-service/config.yaml
STOP_TIMEOUT_SEC=90

View File

@ -0,0 +1,6 @@
# id_generator.node_id in config.yaml must be unique for every active user-service instance.
IMAGE=ccr.ccs.tencentyun.com/hyapp/user-service:REPLACE_WITH_RELEASE_TAG
CONTAINER_NAME=hyapp-user-service
CONFIG_PATH=/etc/hyapp/user-service/config.yaml
STOP_TIMEOUT_SEC=60

View File

@ -0,0 +1,5 @@
IMAGE=ccr.ccs.tencentyun.com/hyapp/wallet-service:REPLACE_WITH_RELEASE_TAG
CONTAINER_NAME=hyapp-wallet-service
CONFIG_PATH=/etc/hyapp/wallet-service/config.yaml
STOP_TIMEOUT_SEC=60

View File

@ -0,0 +1,24 @@
[Unit]
Description=Hyapp activity-service container
Requires=docker.service
After=network-online.target docker.service
Wants=network-online.target
StartLimitIntervalSec=120
StartLimitBurst=3
[Service]
Type=simple
EnvironmentFile=/etc/hyapp/activity-service/docker.env
ExecStartPre=-/usr/bin/docker rm -f ${CONTAINER_NAME}
ExecStart=/usr/bin/docker run --name=${CONTAINER_NAME} --rm --network=host --init --pull=never --stop-signal=SIGTERM --stop-timeout=${STOP_TIMEOUT_SEC} --env=TZ=UTC --log-driver=json-file --log-opt=max-size=100m --log-opt=max-file=5 --mount=type=bind,src=${CONFIG_PATH},dst=/app/config.yaml,readonly --health-cmd="/app/grpc-health-probe -addr=127.0.0.1:13006 -service=activity-service" --health-interval=5s --health-timeout=3s --health-retries=6 --health-start-period=10s ${IMAGE}
ExecStop=/usr/bin/docker stop --time=${STOP_TIMEOUT_SEC} ${CONTAINER_NAME}
Restart=on-failure
RestartSec=5s
TimeoutStartSec=60
TimeoutStopSec=90
KillMode=none
LimitNOFILE=1048576
[Install]
WantedBy=multi-user.target

View File

@ -0,0 +1,24 @@
[Unit]
Description=Hyapp cron-service container
Requires=docker.service
After=network-online.target docker.service
Wants=network-online.target
StartLimitIntervalSec=120
StartLimitBurst=3
[Service]
Type=simple
EnvironmentFile=/etc/hyapp/cron-service/docker.env
ExecStartPre=-/usr/bin/docker rm -f ${CONTAINER_NAME}
ExecStart=/usr/bin/docker run --name=${CONTAINER_NAME} --rm --network=host --init --pull=never --stop-signal=SIGTERM --stop-timeout=${STOP_TIMEOUT_SEC} --env=TZ=UTC --log-driver=json-file --log-opt=max-size=100m --log-opt=max-file=5 --mount=type=bind,src=${CONFIG_PATH},dst=/app/config.yaml,readonly --health-cmd="/app/grpc-health-probe -addr=127.0.0.1:13007 -service=cron-service" --health-interval=5s --health-timeout=3s --health-retries=6 --health-start-period=10s ${IMAGE}
ExecStop=/usr/bin/docker stop --time=${STOP_TIMEOUT_SEC} ${CONTAINER_NAME}
Restart=on-failure
RestartSec=5s
TimeoutStartSec=60
TimeoutStopSec=120
KillMode=none
LimitNOFILE=1048576
[Install]
WantedBy=multi-user.target

View File

@ -0,0 +1,24 @@
[Unit]
Description=Hyapp game-service container
Requires=docker.service
After=network-online.target docker.service
Wants=network-online.target
StartLimitIntervalSec=120
StartLimitBurst=3
[Service]
Type=simple
EnvironmentFile=/etc/hyapp/game-service/docker.env
ExecStartPre=-/usr/bin/docker rm -f ${CONTAINER_NAME}
ExecStart=/usr/bin/docker run --name=${CONTAINER_NAME} --rm --network=host --init --pull=never --stop-signal=SIGTERM --stop-timeout=${STOP_TIMEOUT_SEC} --env=TZ=UTC --log-driver=json-file --log-opt=max-size=100m --log-opt=max-file=5 --mount=type=bind,src=${CONFIG_PATH},dst=/app/config.yaml,readonly --health-cmd="/app/grpc-health-probe -addr=127.0.0.1:13008 -service=game-service" --health-interval=5s --health-timeout=3s --health-retries=6 --health-start-period=10s ${IMAGE}
ExecStop=/usr/bin/docker stop --time=${STOP_TIMEOUT_SEC} ${CONTAINER_NAME}
Restart=on-failure
RestartSec=5s
TimeoutStartSec=60
TimeoutStopSec=90
KillMode=none
LimitNOFILE=1048576
[Install]
WantedBy=multi-user.target

View File

@ -0,0 +1,24 @@
[Unit]
Description=Hyapp gateway-service container
Requires=docker.service
After=network-online.target docker.service
Wants=network-online.target
StartLimitIntervalSec=120
StartLimitBurst=3
[Service]
Type=simple
EnvironmentFile=/etc/hyapp/gateway-service/docker.env
ExecStartPre=-/usr/bin/docker rm -f ${CONTAINER_NAME}
ExecStart=/usr/bin/docker run --name=${CONTAINER_NAME} --rm --network=host --init --pull=never --stop-signal=SIGTERM --stop-timeout=${STOP_TIMEOUT_SEC} --env=TZ=UTC --log-driver=json-file --log-opt=max-size=100m --log-opt=max-file=5 --mount=type=bind,src=${CONFIG_PATH},dst=/app/config.yaml,readonly --health-cmd="wget -q -O - http://127.0.0.1:13000/healthz/ready >/dev/null" --health-interval=5s --health-timeout=3s --health-retries=6 --health-start-period=10s ${IMAGE}
ExecStop=/usr/bin/docker stop --time=${STOP_TIMEOUT_SEC} ${CONTAINER_NAME}
Restart=on-failure
RestartSec=5s
TimeoutStartSec=60
TimeoutStopSec=90
KillMode=none
LimitNOFILE=1048576
[Install]
WantedBy=multi-user.target

View File

@ -0,0 +1,24 @@
[Unit]
Description=Hyapp room-service container
Requires=docker.service
After=network-online.target docker.service
Wants=network-online.target
StartLimitIntervalSec=120
StartLimitBurst=3
[Service]
Type=simple
EnvironmentFile=/etc/hyapp/room-service/docker.env
ExecStartPre=-/usr/bin/docker rm -f ${CONTAINER_NAME}
ExecStart=/usr/bin/docker run --name=${CONTAINER_NAME} --rm --network=host --init --pull=never --stop-signal=SIGTERM --stop-timeout=${STOP_TIMEOUT_SEC} --env=TZ=UTC --log-driver=json-file --log-opt=max-size=100m --log-opt=max-file=5 --mount=type=bind,src=${CONFIG_PATH},dst=/app/config.yaml,readonly --health-cmd="/app/grpc-health-probe -addr=127.0.0.1:13001 -service=room-service" --health-interval=5s --health-timeout=3s --health-retries=6 --health-start-period=10s ${IMAGE}
ExecStop=/usr/bin/docker stop --time=${STOP_TIMEOUT_SEC} ${CONTAINER_NAME}
Restart=on-failure
RestartSec=5s
TimeoutStartSec=60
TimeoutStopSec=120
KillMode=none
LimitNOFILE=1048576
[Install]
WantedBy=multi-user.target

View File

@ -0,0 +1,24 @@
[Unit]
Description=Hyapp user-service container
Requires=docker.service
After=network-online.target docker.service
Wants=network-online.target
StartLimitIntervalSec=120
StartLimitBurst=3
[Service]
Type=simple
EnvironmentFile=/etc/hyapp/user-service/docker.env
ExecStartPre=-/usr/bin/docker rm -f ${CONTAINER_NAME}
ExecStart=/usr/bin/docker run --name=${CONTAINER_NAME} --rm --network=host --init --pull=never --stop-signal=SIGTERM --stop-timeout=${STOP_TIMEOUT_SEC} --env=TZ=UTC --log-driver=json-file --log-opt=max-size=100m --log-opt=max-file=5 --mount=type=bind,src=${CONFIG_PATH},dst=/app/config.yaml,readonly --health-cmd="/app/grpc-health-probe -addr=127.0.0.1:13005 -service=user-service" --health-interval=5s --health-timeout=3s --health-retries=6 --health-start-period=10s ${IMAGE}
ExecStop=/usr/bin/docker stop --time=${STOP_TIMEOUT_SEC} ${CONTAINER_NAME}
Restart=on-failure
RestartSec=5s
TimeoutStartSec=60
TimeoutStopSec=90
KillMode=none
LimitNOFILE=1048576
[Install]
WantedBy=multi-user.target

View File

@ -0,0 +1,24 @@
[Unit]
Description=Hyapp wallet-service container
Requires=docker.service
After=network-online.target docker.service
Wants=network-online.target
StartLimitIntervalSec=120
StartLimitBurst=3
[Service]
Type=simple
EnvironmentFile=/etc/hyapp/wallet-service/docker.env
ExecStartPre=-/usr/bin/docker rm -f ${CONTAINER_NAME}
ExecStart=/usr/bin/docker run --name=${CONTAINER_NAME} --rm --network=host --init --pull=never --stop-signal=SIGTERM --stop-timeout=${STOP_TIMEOUT_SEC} --env=TZ=UTC --log-driver=json-file --log-opt=max-size=100m --log-opt=max-file=5 --mount=type=bind,src=${CONFIG_PATH},dst=/app/config.yaml,readonly --health-cmd="/app/grpc-health-probe -addr=127.0.0.1:13004 -service=wallet-service" --health-interval=5s --health-timeout=3s --health-retries=6 --health-start-period=10s ${IMAGE}
ExecStop=/usr/bin/docker stop --time=${STOP_TIMEOUT_SEC} ${CONTAINER_NAME}
Restart=on-failure
RestartSec=5s
TimeoutStartSec=60
TimeoutStopSec=90
KillMode=none
LimitNOFILE=1048576
[Install]
WantedBy=multi-user.target

View File

@ -39,6 +39,7 @@ services:
TZ: UTC
ports:
- "13001:13001"
- "13101:13101"
depends_on:
mysql:
condition: service_healthy
@ -47,7 +48,7 @@ services:
wallet-service:
condition: service_healthy
healthcheck:
test: ["CMD-SHELL", "/app/grpc-health-probe -addr=127.0.0.1:13001 -service=room-service"]
test: ["CMD-SHELL", "wget -q -O - http://127.0.0.1:13101/healthz/ready >/dev/null"]
interval: 5s
timeout: 3s
retries: 20
@ -63,6 +64,7 @@ services:
TZ: UTC
ports:
- "13004:13004"
- "13104:13104"
depends_on:
mysql:
condition: service_healthy
@ -71,7 +73,7 @@ services:
activity-service:
condition: service_healthy
healthcheck:
test: ["CMD-SHELL", "/app/grpc-health-probe -addr=127.0.0.1:13004 -service=wallet-service"]
test: ["CMD-SHELL", "wget -q -O - http://127.0.0.1:13104/healthz/ready >/dev/null"]
interval: 5s
timeout: 3s
retries: 20
@ -87,11 +89,12 @@ services:
TZ: UTC
ports:
- "13005:13005"
- "13105:13105"
depends_on:
mysql:
condition: service_healthy
healthcheck:
test: ["CMD-SHELL", "/app/grpc-health-probe -addr=127.0.0.1:13005 -service=user-service"]
test: ["CMD-SHELL", "wget -q -O - http://127.0.0.1:13105/healthz/ready >/dev/null"]
interval: 5s
timeout: 3s
retries: 20
@ -107,13 +110,14 @@ services:
TZ: UTC
ports:
- "13006:13006"
- "13106:13106"
depends_on:
mysql:
condition: service_healthy
user-service:
condition: service_healthy
healthcheck:
test: ["CMD-SHELL", "/app/grpc-health-probe -addr=127.0.0.1:13006 -service=activity-service"]
test: ["CMD-SHELL", "wget -q -O - http://127.0.0.1:13106/healthz/ready >/dev/null"]
interval: 5s
timeout: 3s
retries: 20
@ -129,6 +133,7 @@ services:
TZ: UTC
ports:
- "13007:13007"
- "13107:13107"
depends_on:
mysql:
condition: service_healthy
@ -137,7 +142,7 @@ services:
activity-service:
condition: service_healthy
healthcheck:
test: ["CMD-SHELL", "/app/grpc-health-probe -addr=127.0.0.1:13007 -service=cron-service"]
test: ["CMD-SHELL", "wget -q -O - http://127.0.0.1:13107/healthz/ready >/dev/null"]
interval: 5s
timeout: 3s
retries: 20
@ -153,6 +158,7 @@ services:
TZ: UTC
ports:
- "13008:13008"
- "13108:13108"
depends_on:
mysql:
condition: service_healthy
@ -161,7 +167,7 @@ services:
user-service:
condition: service_healthy
healthcheck:
test: ["CMD-SHELL", "/app/grpc-health-probe -addr=127.0.0.1:13008 -service=game-service"]
test: ["CMD-SHELL", "wget -q -O - http://127.0.0.1:13108/healthz/ready >/dev/null"]
interval: 5s
timeout: 3s
retries: 20

View File

@ -126,7 +126,8 @@ SIGINT
- 对已经进入处理中的命令,允许在 deadline 内完成提交。
- 命令提交后必须完成必要 command log、snapshot dirty 标记和 outbox 写入。
- 对本节点持有的 Room Cell 执行 snapshot flush。
- 停止 lease 续约,或按实现安全释放 lease。
- 停止 node registry 心跳gRPC drain 完成后删除 `room:node:{node_id}`
- gRPC drain 完成后释放本节点仍持有的已加载房间 lease让其他节点无需等待 `lease_ttl` 才能接管。
- outbox worker 处理完当前批次,未处理事件保持在 MySQL 中等待其他 worker 或下次启动补投。
Room Cell 下线顺序:
@ -137,7 +138,9 @@ Room Cell 下线顺序:
3. command log/outbox transaction completes
4. snapshot flush runs with timeout
5. lease renew stops
6. cell is removed from local manager
6. owner route is released after in-flight commands finish
7. node registry is removed
8. cell is removed from local manager
```
如果 shutdown timeout 到期:
@ -145,6 +148,7 @@ Room Cell 下线顺序:
- 不能删除未安全释放的持久状态。
- 不能清空 outbox。
- 可以让 Redis lease 自然过期,由其他节点通过 snapshot + command log 恢复。
- 如果主动释放 lease 未完成,不能强删 MySQL 状态;让 Redis TTL 自然过期是安全兜底。
禁止行为:

View File

@ -147,6 +147,7 @@ activity-service
- 当前节点不处于 draining 状态。
- gRPC listener 已启动。
- Redis lease 操作能力正常,至少能执行一次轻量读写或 Lua 脚本探测。
- 节点启动时必须能写入 `room:node:{node_id}`;运行期心跳失败要记录告警,避免其他 room 节点无法转发到当前 owner。
`wallet-service` 建议作为 `SendGift` 链路的强依赖监控,但不要默认作为 `room-service` 整体 ready 的硬失败条件;否则 wallet 短暂异常会导致房间基础命令也被摘流。

View File

@ -0,0 +1,112 @@
# 多机部署与无感发布
本文定义不使用 Kubernetes 时的生产部署边界。目标是让 gateway 和内部微服务可以多机滚动发布,并让 `room-service` 在普通内网负载均衡命中非 owner 节点时自动转发到当前 Room Cell owner。
## 网络拓扑
推荐拓扑:
```text
公网 CLB
-> gateway-service 多实例
-> room-service 内网入口
-> user-service 内网入口
-> wallet-service 内网入口
-> activity-service 内网入口
-> game-service 内网入口
room-service 实例之间通过 Redis owner route 和 node registry 直连转发。
```
规则:
- 公网只暴露 `gateway-service`
- 内部 gRPC 服务只监听私网,安全组只允许服务网段访问 `13xxx` 端口。
- `user-service``wallet-service``activity-service``game-service` 可以直接挂内网 CLB 或 DNS。
- `room-service` 可以挂内网 CLB/DNS 给 gateway 使用,但 owner 转发必须使用实例级 `advertise_addr`,不能使用 CLB 地址。
## Room Owner 路由
`room-service` 使用 Redis 维护两个短租约:
```text
room:route:{app_code + "\0" + room_id} -> owner node_id / lease_token / expires_at_ms
room:node:{node_id} -> 实例 advertise_addr / expires_at_ms
```
写命令、IM guard 和完整房间快照查询的处理顺序:
1. 请求先到任意 `room-service` 节点。
2. 节点按 `app_code + "\0" + room_id` 查询 Redis owner route。
3. 如果当前节点就是 owner直接进入本地 Room Cell。
4. 如果 owner 是其他节点,按 `room:node:{owner_node_id}` 找到实例私网地址并转发 gRPC。
5. 如果 route 不存在或已过期,当前节点按原有 `EnsureOwner` 逻辑接管并从 MySQL snapshot + command log 恢复。
转发只做一跳。`advertise_addr` 写错成 CLB 地址时,请求会在一跳后停止递归,并暴露真实 owner 冲突或 unavailable。
## room-service 配置
每个实例必须有唯一配置:
```yaml
node_id: "room-node-prod-1"
grpc_addr: ":13001"
advertise_addr: "10.0.1.12:13001"
health_http_addr: ":13101"
node_registry_ttl: "30s"
node_registry_heartbeat_interval: "10s"
owner_forward_timeout: "2s"
lease_ttl: "10s"
```
配置边界:
- `grpc_addr` 是本进程监听地址,可以是 `:13001`
- `advertise_addr` 是其他 room 节点能直连的实例私网地址,必须是单实例地址。
- `node_id` 参与 room owner lease、节点注册和日志排障不能在多个实例间复用。
- `node_registry_ttl` 必须大于心跳周期;默认 `30s/10s`
- `lease_ttl` 是节点故障后的自动接管窗口;正常滚动发布会在 gRPC drain 后主动释放本节点已加载房间 lease减少等待 TTL 的窗口。
## 无感发布步骤
单个服务滚动发布顺序:
1. 在新机器或新端口启动新版本实例。
2. 等 `health_http_addr``/readyz` 或 gRPC health 通过。
3. 把新实例加入对应 CLB 后端或内部 DNS。
4. 观察错误率、延迟和日志。
5. 从 CLB 摘除一个旧实例,等待 CLB 健康检查和连接排空周期。
6. 对旧实例发送 `SIGTERM`
7. 旧实例进入 draininggRPC 完成 in-flight 请求后释放已加载房间 lease并删除 `room:node:{node_id}`
8. 新请求命中新节点后,如果 Redis route 已释放,会立即接管并从 MySQL 恢复;如果旧实例异常退出,则等待 `lease_ttl` 到期后接管。
发布约束:
- 不要先杀进程再摘 CLB。
- 不要把 `room-service``advertise_addr` 写成内网 CLB。
- 不要多个实例复用同一个 `node_id`
- 数据库变更要先执行兼容当前发布窗口的迁移,再滚动服务。
## 配置拆分
不用统一配置中心也可以上线。推荐发布系统渲染最终 `config.yaml`
- 公共环境配置MySQL/Redis 私网地址、日志等级、腾讯云 endpoint。
- 服务配置端口、worker 周期、批量大小、超时。
- 实例配置:`node_id``advertise_addr``user-service.id_generator.node_id`
- SecretMySQL/Redis 密码、JWT secret、腾讯 IM/RTC/COS secret、callback token。
只有当需要动态配置、灰度下发、审计回滚时,再引入 Consul/Nacos/Apollo。当前代码的启动配置仍以单服务 YAML 为边界。
## 真实数据验证
本地真实 MySQL/Redis 验证:
```bash
docker compose up -d mysql redis
ROOM_SERVICE_MYSQL_TEST_DSN='hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_room?parseTime=true&charset=utf8mb4&loc=UTC' \
ROOM_SERVICE_REDIS_TEST_ADDR='127.0.0.1:13379' \
go test -count=1 ./services/room-service/internal/transport/grpc -run TestOwnerForwardingUsesRedisRouteAndNodeRegistry
```
该测试会启动两个真实 gRPC `room-service` 实例,共用真实 MySQL schema 和 Redis route/node registry在 node A 创建房间,通过 node B 调 `JoinRoom``CheckSpeakPermission``GetRoomSnapshot`,验证请求会转发到当前 owner。

View File

@ -96,12 +96,15 @@ stateDiagram-v2
规则:
- Redis runtime route key 统一使用 `app_code + "\0" + room_id`,包括默认 App不能再使用裸 `room_id` 作为兼容 key。
- `room:node:{node_id}` 保存当前 room-service 实例的 `advertise_addr`,只用于 owner 转发;它不是服务发现的全局配置中心。
- gateway 可以连接 room-service 内网负载均衡;命中非 owner 节点时gRPC 边界必须按 Redis owner route 转发到当前 owner 实例。
- `advertise_addr` 必须是实例私网地址,不能是 room-service 内网 CLB 地址,否则转发会命中任意节点并触发递归保护。
- 命令开始前通过 `EnsureOwner` 获得执行权。
- 命令真正写 command log/outbox 前必须用 `lease_token` 再做一次 `VerifyOwner`
- 如果提交前 lease 已经过期或 token 已变化,旧节点必须返回 `UNAVAILABLE`,不能写房间状态。
- `room_command_log` 保存 `owner_node_id``lease_token`,用于排查谁在什么 lease 下提交了命令。
- stale presence worker 和 mic publish timeout worker 只能扫描本节点已加载 Cell清理前必须确认当前 Redis lease 仍属于本节点,并在读完快照后再次用相同 `lease_token` 校验,最后由命令提交链路再做 fencing。
- 节点进入 draining 后不能再接收新的 Room Cell 命令,也不能恢复或创建新的 Cell。
- 节点进入 draining 后不能再接收新的 Room Cell 命令,也不能恢复或创建新的 CellgRPC drain 完成后应主动释放本节点已加载房间的 owner lease降低滚动发布等待 TTL 的窗口
## Outbox

166
pkg/grpcclient/client.go Normal file
View File

@ -0,0 +1,166 @@
// Package grpcclient provides one production dial policy for internal gRPC
// clients. The policy keeps transport behavior out of business handlers.
package grpcclient
import (
"context"
"fmt"
"strings"
"time"
"google.golang.org/grpc"
"google.golang.org/grpc/backoff"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/keepalive"
)
// Config controls connection backoff, keepalive, retry and per-RPC deadlines.
type Config struct {
DefaultTimeout time.Duration
ConnectTimeout time.Duration
KeepaliveTime time.Duration
KeepaliveTimeout time.Duration
RetryMaxAttempts int
RetryInitialBackoff time.Duration
RetryMaxBackoff time.Duration
RetryBackoffMultiplier float64
RetryableStatusCodes []string
}
// DefaultConfig is conservative: retries are limited to one retry on
// UNAVAILABLE and must complete inside the caller's deadline.
func DefaultConfig() Config {
return Config{
DefaultTimeout: 5 * time.Second,
ConnectTimeout: 2 * time.Second,
KeepaliveTime: 30 * time.Second,
KeepaliveTimeout: 10 * time.Second,
RetryMaxAttempts: 2,
RetryInitialBackoff: 100 * time.Millisecond,
RetryMaxBackoff: 500 * time.Millisecond,
RetryBackoffMultiplier: 2,
RetryableStatusCodes: []string{"UNAVAILABLE"},
}
}
// Normalize fills safe transport defaults and clamps retry attempts to grpc-go's
// supported range. A value below two disables explicit retry.
func Normalize(cfg Config) Config {
defaults := DefaultConfig()
if cfg.DefaultTimeout <= 0 {
cfg.DefaultTimeout = defaults.DefaultTimeout
}
if cfg.ConnectTimeout <= 0 {
cfg.ConnectTimeout = defaults.ConnectTimeout
}
if cfg.KeepaliveTime <= 0 {
cfg.KeepaliveTime = defaults.KeepaliveTime
}
if cfg.KeepaliveTimeout <= 0 {
cfg.KeepaliveTimeout = defaults.KeepaliveTimeout
}
if cfg.RetryMaxAttempts < 0 {
cfg.RetryMaxAttempts = 0
}
if cfg.RetryMaxAttempts == 0 {
cfg.RetryMaxAttempts = defaults.RetryMaxAttempts
}
if cfg.RetryMaxAttempts > 5 {
cfg.RetryMaxAttempts = 5
}
if cfg.RetryInitialBackoff <= 0 {
cfg.RetryInitialBackoff = defaults.RetryInitialBackoff
}
if cfg.RetryMaxBackoff <= 0 {
cfg.RetryMaxBackoff = defaults.RetryMaxBackoff
}
if cfg.RetryMaxBackoff < cfg.RetryInitialBackoff {
cfg.RetryMaxBackoff = cfg.RetryInitialBackoff
}
if cfg.RetryBackoffMultiplier <= 0 {
cfg.RetryBackoffMultiplier = defaults.RetryBackoffMultiplier
}
if len(cfg.RetryableStatusCodes) == 0 {
cfg.RetryableStatusCodes = defaults.RetryableStatusCodes
}
return cfg
}
// Dial creates a ClientConn with deadlines, TCP reconnect backoff, keepalive and
// a minimal retry policy. Startup does not block on the remote endpoint; ready
// probes are responsible for proving the connection can become READY.
func Dial(target string, cfg Config, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
cfg = Normalize(cfg)
dialOptions := []grpc.DialOption{
grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithConnectParams(grpc.ConnectParams{
Backoff: backoff.Config{
BaseDelay: 100 * time.Millisecond,
Multiplier: 1.6,
Jitter: 0.2,
MaxDelay: 5 * time.Second,
},
MinConnectTimeout: cfg.ConnectTimeout,
}),
grpc.WithKeepaliveParams(keepalive.ClientParameters{
Time: cfg.KeepaliveTime,
Timeout: cfg.KeepaliveTimeout,
PermitWithoutStream: true,
}),
grpc.WithUnaryInterceptor(timeoutUnaryInterceptor(cfg.DefaultTimeout)),
}
if serviceConfig := defaultServiceConfig(cfg); serviceConfig != "" {
// The retry budget is intentionally transport-level and narrow. Business
// handlers still rely on command IDs/idempotency for mutation safety.
dialOptions = append(dialOptions, grpc.WithDefaultServiceConfig(serviceConfig))
}
dialOptions = append(dialOptions, opts...)
return grpc.Dial(strings.TrimSpace(target), dialOptions...)
}
func timeoutUnaryInterceptor(timeout time.Duration) grpc.UnaryClientInterceptor {
return func(ctx context.Context, method string, req any, reply any, conn *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
if timeout > 0 {
if _, ok := ctx.Deadline(); !ok {
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, timeout)
defer cancel()
}
}
return invoker(ctx, method, req, reply, conn, opts...)
}
}
func defaultServiceConfig(cfg Config) string {
if cfg.RetryMaxAttempts < 2 {
return `{"methodConfig":[{"name":[{}],"waitForReady":true}]}`
}
return fmt.Sprintf(
`{"methodConfig":[{"name":[{}],"waitForReady":true,"retryPolicy":{"maxAttempts":%d,"initialBackoff":"%s","maxBackoff":"%s","backoffMultiplier":%.2f,"retryableStatusCodes":[%s]}}]}`,
cfg.RetryMaxAttempts,
durationString(cfg.RetryInitialBackoff),
durationString(cfg.RetryMaxBackoff),
cfg.RetryBackoffMultiplier,
statusCodeList(cfg.RetryableStatusCodes),
)
}
func durationString(value time.Duration) string {
return fmt.Sprintf("%.3fs", value.Seconds())
}
func statusCodeList(codes []string) string {
quoted := make([]string, 0, len(codes))
for _, code := range codes {
code = strings.ToUpper(strings.TrimSpace(code))
if code == "" {
continue
}
quoted = append(quoted, fmt.Sprintf("%q", code))
}
if len(quoted) == 0 {
return `"UNAVAILABLE"`
}
return strings.Join(quoted, ",")
}

View File

@ -0,0 +1,52 @@
package grpcclient
import (
"context"
"strings"
"testing"
"time"
"google.golang.org/grpc"
)
func TestNormalizeRestoresTransportDefaults(t *testing.T) {
cfg := Normalize(Config{})
if cfg.DefaultTimeout != 5*time.Second || cfg.ConnectTimeout != 2*time.Second {
t.Fatalf("deadline defaults mismatch: %+v", cfg)
}
if cfg.KeepaliveTime != 30*time.Second || cfg.KeepaliveTimeout != 10*time.Second {
t.Fatalf("keepalive defaults mismatch: %+v", cfg)
}
if cfg.RetryMaxAttempts != 2 || cfg.RetryInitialBackoff != 100*time.Millisecond || cfg.RetryMaxBackoff != 500*time.Millisecond {
t.Fatalf("retry defaults mismatch: %+v", cfg)
}
}
func TestTimeoutInterceptorPreservesExistingDeadline(t *testing.T) {
parent, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()
before, _ := parent.Deadline()
interceptor := timeoutUnaryInterceptor(time.Millisecond)
err := interceptor(parent, "/svc/Method", nil, nil, nil, func(ctx context.Context, method string, req any, reply any, conn *grpc.ClientConn, opts ...grpc.CallOption) error {
after, ok := ctx.Deadline()
if !ok || !after.Equal(before) {
t.Fatalf("deadline should be preserved: ok=%t before=%s after=%s", ok, before, after)
}
return nil
})
if err != nil {
t.Fatalf("interceptor returned error: %v", err)
}
}
func TestDefaultServiceConfigIncludesUnavailableRetry(t *testing.T) {
serviceConfig := defaultServiceConfig(DefaultConfig())
if !strings.Contains(serviceConfig, `"retryableStatusCodes":["UNAVAILABLE"]`) {
t.Fatalf("retry policy missing UNAVAILABLE: %s", serviceConfig)
}
if !strings.Contains(serviceConfig, `"waitForReady":true`) {
t.Fatalf("service config should enable waitForReady: %s", serviceConfig)
}
}

View File

@ -19,9 +19,9 @@ const (
// Check 描述一项 readiness 子检查gRPC health 只返回总状态,子项用于服务内聚合。
type Check struct {
Name string
Status string
Message string
Name string `json:"name"`
Status string `json:"status"`
Message string `json:"message,omitempty"`
}
// Checker 把服务自己的 ready 语义适配到标准 gRPC health protocol。

View File

@ -2,6 +2,7 @@ package grpchealth
import (
"context"
"errors"
"testing"
healthpb "google.golang.org/grpc/health/grpc_health_v1"
@ -37,3 +38,18 @@ func TestServingCheckerMapsReadiness(t *testing.T) {
t.Fatalf("status after draining mismatch: got %s want NOT_SERVING", resp.GetStatus())
}
}
func TestServingCheckerDependencyFailureAffectsReady(t *testing.T) {
checker := NewServingChecker("wallet-service", Dependency{
Name: "mysql",
Check: func(context.Context) error {
return errors.New("mysql down")
},
})
checker.MarkServing()
checks := checker.Ready(context.Background())
if Healthy(checks) {
t.Fatalf("dependency failure should make checker not ready: %+v", checks)
}
}

View File

@ -2,19 +2,31 @@ package grpchealth
import (
"context"
"strings"
"sync/atomic"
"time"
)
const defaultDependencyBudget = 300 * time.Millisecond
// Dependency 是 readiness 中必须可用的外部依赖。
type Dependency struct {
Name string
Budget time.Duration
Check func(ctx context.Context) error
}
// ServingChecker 覆盖无外部硬依赖的占位或 stub gRPC 服务 readiness。
type ServingChecker struct {
serviceName string
serving atomic.Bool
draining atomic.Bool
serviceName string
dependencies []Dependency
serving atomic.Bool
draining atomic.Bool
}
// NewServingChecker 创建只依赖 gRPC listener 和 draining 状态的 checker。
func NewServingChecker(serviceName string) *ServingChecker {
return &ServingChecker{serviceName: serviceName}
func NewServingChecker(serviceName string, dependencies ...Dependency) *ServingChecker {
return &ServingChecker{serviceName: serviceName, dependencies: dependencies}
}
// ServiceName 返回标准 gRPC health 使用的服务边界名。
@ -37,10 +49,35 @@ func (c *ServingChecker) MarkDraining() {
c.draining.Store(true)
}
// Ready 返回无数据库类硬依赖服务的最小 readiness。
func (c *ServingChecker) Ready(_ context.Context) []Check {
return []Check{
// Ready 返回 listener、draining 和外部硬依赖的聚合 readiness。
func (c *ServingChecker) Ready(ctx context.Context) []Check {
checks := []Check{
Bool("grpc_listener", c.serving.Load(), "gRPC listener is not serving"),
Bool("draining", !c.draining.Load(), "node is draining"),
}
for _, dependency := range c.dependencies {
checks = append(checks, dependency.Ready(ctx))
}
return checks
}
// Ready executes a dependency check with a short bounded budget.
func (d Dependency) Ready(ctx context.Context) Check {
name := strings.TrimSpace(d.Name)
if name == "" {
name = "dependency"
}
if d.Check == nil {
return Fail(name, "dependency check is not configured")
}
budget := d.Budget
if budget <= 0 {
budget = defaultDependencyBudget
}
checkCtx, cancel := context.WithTimeout(ctx, budget)
defer cancel()
if err := d.Check(checkCtx); err != nil {
return Fail(name, err.Error())
}
return OK(name)
}

View File

@ -0,0 +1,34 @@
// Package grpcshutdown centralizes gRPC draining semantics for all services.
package grpcshutdown
import (
"time"
"google.golang.org/grpc"
)
const defaultTimeout = 15 * time.Second
// GracefulStop waits for in-flight RPCs to finish, then falls back to Stop so a
// wedged client stream cannot block deploy shutdown forever.
func GracefulStop(server *grpc.Server, timeout time.Duration) {
if server == nil {
return
}
if timeout <= 0 {
timeout = defaultTimeout
}
done := make(chan struct{})
go func() {
server.GracefulStop()
close(done)
}()
select {
case <-done:
case <-time.After(timeout):
server.Stop()
<-done
}
}

134
pkg/healthhttp/server.go Normal file
View File

@ -0,0 +1,134 @@
// Package healthhttp exposes process-local HTTP health endpoints for services
// whose primary protocol is not HTTP.
package healthhttp
import (
"context"
"encoding/json"
"errors"
"net"
"net/http"
"strings"
"sync"
"time"
"hyapp/pkg/grpchealth"
)
const (
statusOK = "ok"
statusFail = "fail"
)
// LiveChecker allows services with richer runtime liveness checks, such as
// room-service, to reuse their existing process probes.
type LiveChecker interface {
Live() []grpchealth.Check
}
// Response is intentionally close to gateway health responses so CLB probes and
// deploy scripts can consume one stable shape across all services.
type Response struct {
Status string `json:"status"`
Service string `json:"service"`
NodeID string `json:"node_id"`
Checks []grpchealth.Check `json:"checks"`
}
// Server owns the separate HTTP listener used by CLB health checks for gRPC
// processes. It does not replace gRPC health; it makes health visible to CLB.
type Server struct {
nodeID string
checker grpchealth.Checker
server *http.Server
listener net.Listener
once sync.Once
}
// New creates a health HTTP server. Empty addr disables the HTTP health port so
// unit tests and special one-off tools can keep a single listener.
func New(addr string, nodeID string, checker grpchealth.Checker) (*Server, error) {
addr = strings.TrimSpace(addr)
if addr == "" {
return nil, nil
}
if checker == nil {
return nil, errors.New("health checker is required")
}
listener, err := net.Listen("tcp", addr)
if err != nil {
return nil, err
}
s := &Server{
nodeID: strings.TrimSpace(nodeID),
checker: checker,
listener: listener,
}
mux := http.NewServeMux()
mux.HandleFunc("/healthz/live", s.live)
mux.HandleFunc("/healthz/ready", s.ready)
s.server = &http.Server{
Handler: mux,
ReadHeaderTimeout: 2 * time.Second,
}
return s, nil
}
// Run serves health endpoints until Close is called or the listener fails.
func (s *Server) Run() error {
if s == nil {
return nil
}
err := s.server.Serve(s.listener)
if errors.Is(err, http.ErrServerClosed) {
return nil
}
return err
}
// Close stops the health endpoint. Readiness should already be failing before
// this is called because the owning service marks its checker draining first.
func (s *Server) Close(ctx context.Context) error {
if s == nil {
return nil
}
var err error
s.once.Do(func() {
err = s.server.Shutdown(ctx)
})
return err
}
func (s *Server) live(writer http.ResponseWriter, request *http.Request) {
checks := []grpchealth.Check{grpchealth.OK("process")}
if live, ok := s.checker.(LiveChecker); ok {
// Services with process-local invariants should expose them on live, but
// live still avoids slow external dependency probes.
checks = live.Live()
}
s.write(writer, checks)
}
func (s *Server) ready(writer http.ResponseWriter, request *http.Request) {
s.write(writer, s.checker.Ready(request.Context()))
}
func (s *Server) write(writer http.ResponseWriter, checks []grpchealth.Check) {
status := statusFail
code := http.StatusServiceUnavailable
if grpchealth.Healthy(checks) {
status = statusOK
code = http.StatusOK
}
writer.Header().Set("Content-Type", "application/json")
writer.WriteHeader(code)
_ = json.NewEncoder(writer).Encode(Response{
Status: status,
Service: s.checker.ServiceName(),
NodeID: s.nodeID,
Checks: checks,
})
}

View File

@ -0,0 +1,73 @@
package healthhttp
import (
"encoding/json"
"net"
"net/http"
"testing"
"hyapp/pkg/grpchealth"
)
func TestHTTPReadyReflectsCheckerState(t *testing.T) {
checker := grpchealth.NewServingChecker("test-service")
server, err := New("127.0.0.1:0", "test-node", checker)
if err != nil {
t.Fatalf("New failed: %v", err)
}
go func() {
_ = server.Run()
}()
t.Cleanup(func() {
_ = server.Close(t.Context())
})
baseURL := "http://" + server.listener.Addr().String()
resp, err := http.Get(baseURL + "/healthz/ready")
if err != nil {
t.Fatalf("GET not-ready failed: %v", err)
}
if resp.StatusCode != http.StatusServiceUnavailable {
t.Fatalf("not-ready status mismatch: got %d", resp.StatusCode)
}
_ = resp.Body.Close()
checker.MarkServing()
resp, err = http.Get(baseURL + "/healthz/ready")
if err != nil {
t.Fatalf("GET ready failed: %v", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Fatalf("ready status mismatch: got %d", resp.StatusCode)
}
var body Response
if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
t.Fatalf("decode response failed: %v", err)
}
if body.Service != "test-service" || body.NodeID != "test-node" || body.Status != "ok" {
t.Fatalf("unexpected response: %+v", body)
}
}
func TestNewDisabledWhenAddrEmpty(t *testing.T) {
server, err := New("", "node", grpchealth.NewServingChecker("svc"))
if err != nil {
t.Fatalf("New empty addr failed: %v", err)
}
if server != nil {
t.Fatalf("empty addr should disable health HTTP server")
}
}
func TestNewFailsWhenPortAlreadyBound(t *testing.T) {
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen failed: %v", err)
}
defer listener.Close()
if _, err := New(listener.Addr().String(), "node", grpchealth.NewServingChecker("svc")); err == nil {
t.Fatalf("New should fail on occupied health port")
}
}

View File

@ -30,7 +30,7 @@ COPY services/activity-service/configs/config.docker.yaml /app/config.yaml
USER appuser
EXPOSE 13006
EXPOSE 13006 13106
ENTRYPOINT ["/app/server"]
CMD ["-config", "/app/config.yaml"]

View File

@ -8,6 +8,7 @@ log:
include_response_body: false
max_payload_bytes: 2048
grpc_addr: ":13006"
health_http_addr: ":13106"
mysql_dsn: "hyapp:hyapp@tcp(mysql:3306)/hyapp_activity?parseTime=true&charset=utf8mb4&loc=UTC"
user_service_addr: "user-service:13005"
wallet_service_addr: "wallet-service:13004"

View File

@ -8,6 +8,7 @@ log:
include_response_body: false
max_payload_bytes: 2048
grpc_addr: ":13006"
health_http_addr: ":13106"
mysql_dsn: "USER:PASSWORD@tcp(TENCENT_CDB_HOST:3306)/hyapp_activity?parseTime=true&charset=utf8mb4&loc=UTC&tls=false"
user_service_addr: "user-service.internal:13005"
wallet_service_addr: "wallet-service.internal:13004"

View File

@ -8,6 +8,7 @@ log:
include_response_body: false
max_payload_bytes: 2048
grpc_addr: ":13006"
health_http_addr: ":13106"
mysql_dsn: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_activity?parseTime=true&charset=utf8mb4&loc=UTC"
user_service_addr: "127.0.0.1:13005"
wallet_service_addr: "127.0.0.1:13004"

View File

@ -13,6 +13,8 @@ import (
activityv1 "hyapp.local/api/proto/activity/v1"
walletv1 "hyapp.local/api/proto/wallet/v1"
"hyapp/pkg/grpchealth"
"hyapp/pkg/grpcshutdown"
"hyapp/pkg/healthhttp"
"hyapp/pkg/logx"
"hyapp/pkg/tencentim"
"hyapp/services/activity-service/internal/client"
@ -30,6 +32,7 @@ type App struct {
server *grpc.Server
listener net.Listener
health *grpchealth.ServingChecker
healthHTTP *healthhttp.Server
mysqlRepo *mysqlstorage.Repository
broadcast *broadcastservice.Service
broadcastWorkerEnabled bool
@ -111,14 +114,26 @@ func New(cfg config.Config) (*App, error) {
broadcastServer := grpcserver.NewBroadcastServer(broadcastSvc)
activityv1.RegisterBroadcastServiceServer(server, broadcastServer)
activityv1.RegisterRoomEventConsumerServiceServer(server, broadcastServer)
health := grpchealth.NewServingChecker("activity-service")
health := grpchealth.NewServingChecker("activity-service", grpchealth.Dependency{
Name: "mysql",
Check: repository.Ping,
})
healthgrpc.RegisterHealthServer(server, grpchealth.NewServer(health))
healthHTTP, err := healthhttp.New(cfg.HealthHTTPAddr, cfg.NodeID, health)
if err != nil {
_ = walletConn.Close()
_ = userConn.Close()
_ = listener.Close()
_ = repository.Close()
return nil, err
}
workerCtx, workerStop := context.WithCancel(context.Background())
return &App{
server: server,
listener: listener,
health: health,
healthHTTP: healthHTTP,
mysqlRepo: repository,
broadcast: broadcastSvc,
broadcastWorkerEnabled: cfg.Broadcast.Enabled,
@ -131,6 +146,7 @@ func New(cfg config.Config) (*App, error) {
// Run 启动 gRPC 服务。
func (a *App) Run() error {
a.runHealthHTTP()
a.health.MarkServing()
if a.broadcastWorkerEnabled && a.broadcast != nil && a.workerCtx != nil {
a.workerWG.Go(func() {
@ -158,7 +174,8 @@ func (a *App) Close() {
a.workerStop()
}
a.workerWG.Wait()
a.server.GracefulStop()
grpcshutdown.GracefulStop(a.server, 15*time.Second)
a.closeHealthHTTP()
if a.userConn != nil {
_ = a.userConn.Close()
}
@ -171,3 +188,23 @@ func (a *App) Close() {
}
})
}
func (a *App) runHealthHTTP() {
if a.healthHTTP == nil {
return
}
go func() {
if err := a.healthHTTP.Run(); err != nil {
logx.Error(context.Background(), "health_http_run_failed", err)
}
}()
}
func (a *App) closeHealthHTTP() {
if a.healthHTTP == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
_ = a.healthHTTP.Close(ctx)
}

View File

@ -16,6 +16,8 @@ type Config struct {
NodeID string `yaml:"node_id"`
Environment string `yaml:"environment"`
GRPCAddr string `yaml:"grpc_addr"`
// HealthHTTPAddr 是给 CLB/发布脚本探活使用的 HTTP 监听地址,不承载业务流量。
HealthHTTPAddr string `yaml:"health_http_addr"`
// MySQLDSN 是活动状态、事件消费幂等和 outbox 的必需存储连接串。
MySQLDSN string `yaml:"mysql_dsn"`
// UserServiceAddr 是 fanout worker 读取目标用户游标的内部 gRPC 地址。
@ -91,6 +93,7 @@ func Default() Config {
NodeID: "activity-local",
Environment: "local",
GRPCAddr: ":13006",
HealthHTTPAddr: ":13106",
MySQLDSN: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_activity?parseTime=true&charset=utf8mb4&loc=UTC",
UserServiceAddr: "127.0.0.1:13005",
WalletServiceAddr: "127.0.0.1:13004",
@ -155,6 +158,10 @@ func Load(path string) (Config, error) {
if cfg.Environment == "" {
cfg.Environment = "local"
}
cfg.HealthHTTPAddr = strings.TrimSpace(cfg.HealthHTTPAddr)
if cfg.HealthHTTPAddr == "" {
cfg.HealthHTTPAddr = ":13106"
}
cfg.TencentIM.AdminIdentifier = strings.TrimSpace(cfg.TencentIM.AdminIdentifier)
if cfg.TencentIM.AdminIdentifier == "" {
cfg.TencentIM.AdminIdentifier = "administrator"

View File

@ -30,7 +30,7 @@ COPY services/cron-service/configs/config.docker.yaml /app/config.yaml
USER appuser
EXPOSE 13007
EXPOSE 13007 13107
ENTRYPOINT ["/app/server"]
CMD ["-config", "/app/config.yaml"]

View File

@ -8,6 +8,7 @@ log:
include_response_body: false
max_payload_bytes: 2048
grpc_addr: ":13007"
health_http_addr: ":13107"
mysql_dsn: "hyapp:hyapp@tcp(mysql:3306)/hyapp_cron?parseTime=true&charset=utf8mb4&loc=UTC"
user_service_addr: "user-service:13005"
activity_service_addr: "activity-service:13006"

View File

@ -8,6 +8,7 @@ log:
include_response_body: false
max_payload_bytes: 2048
grpc_addr: ":13007"
health_http_addr: ":13107"
mysql_dsn: "USER:PASSWORD@tcp(TENCENT_CDB_HOST:3306)/hyapp_cron?parseTime=true&charset=utf8mb4&loc=UTC&tls=false"
user_service_addr: "user-service.internal:13005"
activity_service_addr: "activity-service.internal:13006"

View File

@ -8,6 +8,7 @@ log:
include_response_body: false
max_payload_bytes: 2048
grpc_addr: ":13007"
health_http_addr: ":13107"
mysql_dsn: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_cron?parseTime=true&charset=utf8mb4&loc=UTC"
user_service_addr: "127.0.0.1:13005"
activity_service_addr: "127.0.0.1:13006"

View File

@ -14,6 +14,8 @@ import (
activityv1 "hyapp.local/api/proto/activity/v1"
userv1 "hyapp.local/api/proto/user/v1"
"hyapp/pkg/grpchealth"
"hyapp/pkg/grpcshutdown"
"hyapp/pkg/healthhttp"
"hyapp/pkg/logx"
"hyapp/services/cron-service/internal/config"
"hyapp/services/cron-service/internal/integration"
@ -26,6 +28,7 @@ type App struct {
server *grpc.Server
listener net.Listener
health *grpchealth.ServingChecker
healthHTTP *healthhttp.Server
repository *mysqlstorage.Repository
scheduler *scheduler.Scheduler
userConn *grpc.ClientConn
@ -67,8 +70,19 @@ func New(cfg config.Config) (*App, error) {
}
server := grpc.NewServer(grpc.UnaryInterceptor(logx.UnaryServerInterceptor("cron-service")))
health := grpchealth.NewServingChecker("cron-service")
health := grpchealth.NewServingChecker("cron-service", grpchealth.Dependency{
Name: "mysql",
Check: repository.Ping,
})
healthgrpc.RegisterHealthServer(server, grpchealth.NewServer(health))
healthHTTP, err := healthhttp.New(cfg.HealthHTTPAddr, cfg.NodeID, health)
if err != nil {
_ = activityConn.Close()
_ = userConn.Close()
_ = listener.Close()
_ = repository.Close()
return nil, err
}
workerCtx, workerCancel := context.WithCancel(context.Background())
userCron := integration.NewUserCronClient(userv1.NewUserCronServiceClient(userConn))
@ -84,6 +98,7 @@ func New(cfg config.Config) (*App, error) {
server: server,
listener: listener,
health: health,
healthHTTP: healthHTTP,
repository: repository,
scheduler: taskScheduler,
userConn: userConn,
@ -95,6 +110,7 @@ func New(cfg config.Config) (*App, error) {
// Run starts scheduler loops and gRPC health serving.
func (a *App) Run() error {
a.runHealthHTTP()
a.health.MarkServing()
defer func() {
if a.workerCancel != nil {
@ -124,7 +140,8 @@ func (a *App) Close() {
a.workerCancel()
}
a.workerWG.Wait()
a.server.GracefulStop()
grpcshutdown.GracefulStop(a.server, 15*time.Second)
a.closeHealthHTTP()
if a.repository != nil {
_ = a.repository.Close()
}
@ -136,3 +153,23 @@ func (a *App) Close() {
}
})
}
func (a *App) runHealthHTTP() {
if a.healthHTTP == nil {
return
}
go func() {
if err := a.healthHTTP.Run(); err != nil {
logx.Error(context.Background(), "health_http_run_failed", err)
}
}()
}
func (a *App) closeHealthHTTP() {
if a.healthHTTP == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
_ = a.healthHTTP.Close(ctx)
}

View File

@ -21,6 +21,8 @@ type Config struct {
Environment string `yaml:"environment"`
// GRPCAddr exposes gRPC health and future internal management APIs.
GRPCAddr string `yaml:"grpc_addr"`
// HealthHTTPAddr exposes HTTP health for CLB and deploy scripts only.
HealthHTTPAddr string `yaml:"health_http_addr"`
// MySQLDSN stores cron run history and schedule-level leases only.
MySQLDSN string `yaml:"mysql_dsn"`
// UserServiceAddr is the owner endpoint for user/auth/host batch tasks.
@ -62,6 +64,7 @@ func Default() Config {
NodeID: "cron-local",
Environment: "local",
GRPCAddr: ":13007",
HealthHTTPAddr: ":13107",
MySQLDSN: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_cron?parseTime=true&charset=utf8mb4&loc=UTC",
UserServiceAddr: "127.0.0.1:13005",
ActivityServiceAddr: "127.0.0.1:13006",
@ -105,6 +108,10 @@ func normalize(cfg *Config) {
if cfg.GRPCAddr == "" {
cfg.GRPCAddr = ":13007"
}
cfg.HealthHTTPAddr = strings.TrimSpace(cfg.HealthHTTPAddr)
if cfg.HealthHTTPAddr == "" {
cfg.HealthHTTPAddr = ":13107"
}
cfg.Tasks = normalizeTasks(cfg.Tasks)
}

View File

@ -20,5 +20,5 @@ COPY --from=builder /out/server /app/server
COPY --from=builder /out/grpc-health-probe /app/grpc-health-probe
COPY services/game-service/configs/config.docker.yaml /app/config.yaml
USER appuser
EXPOSE 13008
EXPOSE 13008 13108
ENTRYPOINT ["/app/server", "-config", "/app/config.yaml"]

View File

@ -2,6 +2,7 @@ service_name: game-service
node_id: game-docker
environment: local
grpc_addr: ":13008"
health_http_addr: ":13108"
mysql_dsn: "hyapp:hyapp@tcp(mysql:3306)/hyapp_game?parseTime=true&charset=utf8mb4&loc=UTC&multiStatements=true"
wallet_service_addr: "wallet-service:13004"
user_service_addr: "user-service:13005"

View File

@ -2,6 +2,7 @@ service_name: game-service
node_id: game-prod-1
environment: prod
grpc_addr: ":13008"
health_http_addr: ":13108"
mysql_dsn: "${GAME_MYSQL_DSN}"
wallet_service_addr: "wallet-service:13004"
user_service_addr: "user-service:13005"

View File

@ -2,6 +2,7 @@ service_name: game-service
node_id: game-local
environment: local
grpc_addr: ":13008"
health_http_addr: ":13108"
mysql_dsn: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_game?parseTime=true&charset=utf8mb4&loc=UTC&multiStatements=true"
wallet_service_addr: "127.0.0.1:13004"
user_service_addr: "127.0.0.1:13005"

View File

@ -12,6 +12,8 @@ import (
healthgrpc "google.golang.org/grpc/health/grpc_health_v1"
gamev1 "hyapp.local/api/proto/game/v1"
"hyapp/pkg/grpchealth"
"hyapp/pkg/grpcshutdown"
"hyapp/pkg/healthhttp"
"hyapp/pkg/logx"
"hyapp/services/game-service/internal/client"
"hyapp/services/game-service/internal/config"
@ -25,6 +27,7 @@ type App struct {
server *grpc.Server
listener net.Listener
health *grpchealth.ServingChecker
healthHTTP *healthhttp.Server
repo *mysqlstorage.Repository
walletConn *grpc.ClientConn
userConn *grpc.ClientConn
@ -64,13 +67,25 @@ func New(cfg config.Config) (*App, error) {
gamev1.RegisterGameAppServiceServer(server, transport)
gamev1.RegisterGameCallbackServiceServer(server, transport)
gamev1.RegisterGameAdminServiceServer(server, transport)
health := grpchealth.NewServingChecker("game-service")
health := grpchealth.NewServingChecker("game-service", grpchealth.Dependency{
Name: "mysql",
Check: repo.Ping,
})
healthgrpc.RegisterHealthServer(server, grpchealth.NewServer(health))
healthHTTP, err := healthhttp.New(cfg.HealthHTTPAddr, cfg.NodeID, health)
if err != nil {
_ = listener.Close()
_ = userConn.Close()
_ = walletConn.Close()
_ = repo.Close()
return nil, err
}
return &App{server: server, listener: listener, health: health, repo: repo, walletConn: walletConn, userConn: userConn}, nil
return &App{server: server, listener: listener, health: health, healthHTTP: healthHTTP, repo: repo, walletConn: walletConn, userConn: userConn}, nil
}
func (a *App) Run() error {
a.runHealthHTTP()
a.health.MarkServing()
err := a.server.Serve(a.listener)
a.health.MarkStopped()
@ -83,7 +98,8 @@ func (a *App) Run() error {
func (a *App) Close() {
a.closeOnce.Do(func() {
a.health.MarkDraining()
a.server.GracefulStop()
grpcshutdown.GracefulStop(a.server, 15*time.Second)
a.closeHealthHTTP()
if a.walletConn != nil {
_ = a.walletConn.Close()
}
@ -95,3 +111,23 @@ func (a *App) Close() {
}
})
}
func (a *App) runHealthHTTP() {
if a.healthHTTP == nil {
return
}
go func() {
if err := a.healthHTTP.Run(); err != nil {
logx.Error(context.Background(), "health_http_run_failed", err)
}
}()
}
func (a *App) closeHealthHTTP() {
if a.healthHTTP == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
_ = a.healthHTTP.Close(ctx)
}

View File

@ -11,10 +11,12 @@ import (
// Config 描述 game-service 启动所需的最小依赖。
type Config struct {
ServiceName string `yaml:"service_name"`
NodeID string `yaml:"node_id"`
Environment string `yaml:"environment"`
GRPCAddr string `yaml:"grpc_addr"`
ServiceName string `yaml:"service_name"`
NodeID string `yaml:"node_id"`
Environment string `yaml:"environment"`
GRPCAddr string `yaml:"grpc_addr"`
// HealthHTTPAddr 是给 CLB/发布脚本探活使用的 HTTP 监听地址,不承载业务流量。
HealthHTTPAddr string `yaml:"health_http_addr"`
MySQLDSN string `yaml:"mysql_dsn"`
WalletServiceAddr string `yaml:"wallet_service_addr"`
UserServiceAddr string `yaml:"user_service_addr"`
@ -29,6 +31,7 @@ func Default() Config {
NodeID: "game-local",
Environment: "local",
GRPCAddr: ":13008",
HealthHTTPAddr: ":13108",
MySQLDSN: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_game?parseTime=true&charset=utf8mb4&loc=UTC&multiStatements=true",
WalletServiceAddr: "127.0.0.1:13004",
UserServiceAddr: "127.0.0.1:13005",
@ -73,6 +76,10 @@ func (cfg *Config) Normalize() error {
if cfg.GRPCAddr == "" {
cfg.GRPCAddr = ":13008"
}
cfg.HealthHTTPAddr = strings.TrimSpace(cfg.HealthHTTPAddr)
if cfg.HealthHTTPAddr == "" {
cfg.HealthHTTPAddr = ":13108"
}
cfg.MySQLDSN = strings.TrimSpace(cfg.MySQLDSN)
if cfg.MySQLDSN == "" {
return fmt.Errorf("mysql_dsn is required")

View File

@ -14,6 +14,17 @@ user_service_addr: "user-service:13005"
wallet_service_addr: "wallet-service:13004"
activity_service_addr: "activity-service:13006"
game_service_addr: "game-service:13008"
grpc_client:
# Docker 本地也使用生产同形态的 deadline、keepalive 和窄重试策略。
default_timeout: "5s"
connect_timeout: "2s"
keepalive_time: "30s"
keepalive_timeout: "10s"
retry_max_attempts: 2
retry_initial_backoff: "100ms"
retry_max_backoff: "500ms"
retry_backoff_multiplier: 2
retryable_status_codes: ["UNAVAILABLE"]
app_config:
# H5 地址由后台 APP配置/H5配置 写入 hyapp_admin.admin_app_configsgateway 只读下发给 App。
mysql_dsn: "hyapp:hyapp@tcp(mysql:3306)/hyapp_admin?parseTime=true&charset=utf8mb4&loc=UTC"

View File

@ -13,6 +13,18 @@ room_service_addr: "room-service.internal:13001"
user_service_addr: "user-service.internal:13005"
wallet_service_addr: "wallet-service.internal:13004"
activity_service_addr: "activity-service.internal:13006"
game_service_addr: "game-service.internal:13008"
grpc_client:
# 线上通过内网 CLB/DNS 访问内部服务;重试只覆盖瞬时 UNAVAILABLE避免放大业务副作用。
default_timeout: "5s"
connect_timeout: "2s"
keepalive_time: "30s"
keepalive_timeout: "10s"
retry_max_attempts: 2
retry_initial_backoff: "100ms"
retry_max_backoff: "500ms"
retry_backoff_multiplier: 2
retryable_status_codes: ["UNAVAILABLE"]
app_config:
# H5 地址由后台 APP配置/H5配置 写入 hyapp_admin.admin_app_configsgateway 只读下发给 App。
mysql_dsn: "TENCENT_MYSQL_USER:TENCENT_MYSQL_PASSWORD@tcp(TENCENT_MYSQL_HOST:3306)/hyapp_admin?parseTime=true&charset=utf8mb4&loc=UTC"

View File

@ -14,6 +14,17 @@ user_service_addr: "127.0.0.1:13005"
wallet_service_addr: "127.0.0.1:13004"
activity_service_addr: "127.0.0.1:13006"
game_service_addr: "127.0.0.1:13008"
grpc_client:
# gateway 内部 gRPC 默认 deadline 包含一次 UNAVAILABLE 重试;业务长调用应在 handler 上下文显式收紧或放宽。
default_timeout: "5s"
connect_timeout: "2s"
keepalive_time: "30s"
keepalive_timeout: "10s"
retry_max_attempts: 2
retry_initial_backoff: "100ms"
retry_max_backoff: "500ms"
retry_backoff_multiplier: 2
retryable_status_codes: ["UNAVAILABLE"]
app_config:
# H5 地址由后台 APP配置/H5配置 写入 hyapp_admin.admin_app_configsgateway 只读下发给 App。
mysql_dsn: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_admin?parseTime=true&charset=utf8mb4&loc=UTC"

View File

@ -12,7 +12,7 @@ import (
"time"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
"hyapp/pkg/grpcclient"
"hyapp/pkg/tencentcos"
"hyapp/services/gateway-service/internal/appconfig"
"hyapp/services/gateway-service/internal/auth"
@ -43,31 +43,32 @@ func New(cfg config.Config) (*App, error) {
return nil, err
}
roomConn, err := grpc.Dial(cfg.RoomServiceAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
grpcConfig := cfg.GRPCClient.Runtime()
roomConn, err := grpcclient.Dial(cfg.RoomServiceAddr, grpcConfig)
if err != nil {
return nil, err
}
userConn, err := grpc.Dial(cfg.UserServiceAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
userConn, err := grpcclient.Dial(cfg.UserServiceAddr, grpcConfig)
if err != nil {
_ = roomConn.Close()
return nil, err
}
walletConn, err := grpc.Dial(cfg.WalletServiceAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
walletConn, err := grpcclient.Dial(cfg.WalletServiceAddr, grpcConfig)
if err != nil {
_ = roomConn.Close()
_ = userConn.Close()
return nil, err
}
activityConn, err := grpc.Dial(cfg.ActivityServiceAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
activityConn, err := grpcclient.Dial(cfg.ActivityServiceAddr, grpcConfig)
if err != nil {
_ = roomConn.Close()
_ = userConn.Close()
_ = walletConn.Close()
return nil, err
}
gameConn, err := grpc.Dial(cfg.GameServiceAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
gameConn, err := grpcclient.Dial(cfg.GameServiceAddr, grpcConfig)
if err != nil {
_ = roomConn.Close()
_ = userConn.Close()
@ -203,7 +204,12 @@ func New(cfg config.Config) (*App, error) {
}
redisClose = joinClose(redisClose, loginRiskRedisClose)
verifier := auth.NewVerifier(cfg.JWTSecret)
healthState := healthcheck.NewState(nodeID(), cfg.JWTSecret, roomConn)
healthState := healthcheck.NewState(nodeID(), cfg.JWTSecret, roomConn,
healthcheck.GRPCDependency{Name: "user_grpc", Conn: userConn},
healthcheck.GRPCDependency{Name: "wallet_grpc", Conn: walletConn},
healthcheck.GRPCDependency{Name: "activity_grpc", Conn: activityConn},
healthcheck.GRPCDependency{Name: "game_grpc", Conn: gameConn},
)
mux := http.NewServeMux()
healthHandler := healthcheck.NewHTTPHandler(healthState)
@ -328,6 +334,14 @@ func (a *App) Close() error {
var err error
a.closeOnce.Do(func() {
a.health.MarkDraining()
shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
// HTTP Shutdown drains accepted requests before client connections close,
// so in-flight handlers can finish their internal gRPC calls.
if shutdownErr := a.server.Shutdown(shutdownCtx); shutdownErr != nil {
// 超过 drain 预算后强制关闭,避免发布脚本被异常长连接卡住。
err = errors.Join(err, shutdownErr, a.server.Close())
}
if a.roomConn != nil {
err = errors.Join(err, a.roomConn.Close())
}
@ -349,8 +363,6 @@ func (a *App) Close() error {
if a.redisClose != nil {
err = errors.Join(err, a.redisClose())
}
err = errors.Join(err, a.server.Close())
})
return err

View File

@ -6,6 +6,7 @@ import (
"time"
"hyapp/pkg/configx"
"hyapp/pkg/grpcclient"
"hyapp/pkg/logx"
"hyapp/pkg/tencentrtc"
)
@ -22,6 +23,7 @@ type Config struct {
WalletServiceAddr string `yaml:"wallet_service_addr"`
ActivityServiceAddr string `yaml:"activity_service_addr"`
GameServiceAddr string `yaml:"game_service_addr"`
GRPCClient GRPCClientConfig `yaml:"grpc_client"`
AuthRateLimit AuthRateLimitConfig `yaml:"auth_rate_limit"`
LoginRisk LoginRiskConfig `yaml:"login_risk"`
AppConfig AppConfigConfig `yaml:"app_config"`
@ -31,6 +33,43 @@ type Config struct {
Log logx.Config `yaml:"log"`
}
// GRPCClientConfig 控制 gateway 访问内部服务的 gRPC transport 策略。
type GRPCClientConfig struct {
// DefaultTimeout 是没有显式 request deadline 时的 RPC 总预算,包含重试耗时。
DefaultTimeout time.Duration `yaml:"default_timeout"`
// ConnectTimeout 是单次建连最小预算,避免内网目标不可达时长期卡住。
ConnectTimeout time.Duration `yaml:"connect_timeout"`
// KeepaliveTime 是空闲连接探活周期,帮助 CLB/NAT 后的半开连接尽早发现。
KeepaliveTime time.Duration `yaml:"keepalive_time"`
// KeepaliveTimeout 是 keepalive ack 等待时间。
KeepaliveTimeout time.Duration `yaml:"keepalive_timeout"`
// RetryMaxAttempts 是包含首次调用在内的最大尝试次数;小于 2 时关闭显式重试。
RetryMaxAttempts int `yaml:"retry_max_attempts"`
// RetryInitialBackoff 是第一次重试前等待时间。
RetryInitialBackoff time.Duration `yaml:"retry_initial_backoff"`
// RetryMaxBackoff 是重试退避上限。
RetryMaxBackoff time.Duration `yaml:"retry_max_backoff"`
// RetryBackoffMultiplier 控制退避增长倍率。
RetryBackoffMultiplier float64 `yaml:"retry_backoff_multiplier"`
// RetryableStatusCodes 限定可重试状态,生产默认只允许 UNAVAILABLE。
RetryableStatusCodes []string `yaml:"retryable_status_codes"`
}
// Runtime 转换为共享 gRPC client 配置,避免 app 层重复维护默认值。
func (c GRPCClientConfig) Runtime() grpcclient.Config {
return grpcclient.Normalize(grpcclient.Config{
DefaultTimeout: c.DefaultTimeout,
ConnectTimeout: c.ConnectTimeout,
KeepaliveTime: c.KeepaliveTime,
KeepaliveTimeout: c.KeepaliveTimeout,
RetryMaxAttempts: c.RetryMaxAttempts,
RetryInitialBackoff: c.RetryInitialBackoff,
RetryMaxBackoff: c.RetryMaxBackoff,
RetryBackoffMultiplier: c.RetryBackoffMultiplier,
RetryableStatusCodes: c.RetryableStatusCodes,
})
}
// LoginRiskConfig 控制 gateway 登录入口快拦和 Redis 风控缓存读取。
type LoginRiskConfig struct {
Enabled bool `yaml:"enabled"`
@ -147,6 +186,17 @@ func Default() Config {
WalletServiceAddr: "127.0.0.1:13004",
ActivityServiceAddr: "127.0.0.1:13006",
GameServiceAddr: "127.0.0.1:13008",
GRPCClient: GRPCClientConfig{
DefaultTimeout: 5 * time.Second,
ConnectTimeout: 2 * time.Second,
KeepaliveTime: 30 * time.Second,
KeepaliveTimeout: 10 * time.Second,
RetryMaxAttempts: 2,
RetryInitialBackoff: 100 * time.Millisecond,
RetryMaxBackoff: 500 * time.Millisecond,
RetryBackoffMultiplier: 2,
RetryableStatusCodes: []string{"UNAVAILABLE"},
},
AppConfig: AppConfigConfig{
MySQLDSN: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_admin?parseTime=true&charset=utf8mb4&loc=UTC",
},
@ -234,6 +284,18 @@ func (cfg *Config) Normalize() error {
if cfg.GameServiceAddr == "" {
cfg.GameServiceAddr = "127.0.0.1:13008"
}
runtimeGRPC := cfg.GRPCClient.Runtime()
cfg.GRPCClient = GRPCClientConfig{
DefaultTimeout: runtimeGRPC.DefaultTimeout,
ConnectTimeout: runtimeGRPC.ConnectTimeout,
KeepaliveTime: runtimeGRPC.KeepaliveTime,
KeepaliveTimeout: runtimeGRPC.KeepaliveTimeout,
RetryMaxAttempts: runtimeGRPC.RetryMaxAttempts,
RetryInitialBackoff: runtimeGRPC.RetryInitialBackoff,
RetryMaxBackoff: runtimeGRPC.RetryMaxBackoff,
RetryBackoffMultiplier: runtimeGRPC.RetryBackoffMultiplier,
RetryableStatusCodes: runtimeGRPC.RetryableStatusCodes,
}
cfg.AppConfig.MySQLDSN = strings.TrimSpace(cfg.AppConfig.MySQLDSN)
cfg.LoginRisk.RedisAddr = strings.TrimSpace(cfg.LoginRisk.RedisAddr)
cfg.LoginRisk.KeyPrefix = strings.TrimSpace(cfg.LoginRisk.KeyPrefix)

View File

@ -22,6 +22,9 @@ func TestLoad(t *testing.T) {
if cfg.RoomServiceAddr != "127.0.0.1:13001" {
t.Fatalf("unexpected room addr: %s", cfg.RoomServiceAddr)
}
if cfg.GRPCClient.DefaultTimeout != 5*time.Second || cfg.GRPCClient.RetryMaxAttempts != 2 || len(cfg.GRPCClient.RetryableStatusCodes) != 1 || cfg.GRPCClient.RetryableStatusCodes[0] != "UNAVAILABLE" {
t.Fatalf("unexpected grpc client policy: %+v", cfg.GRPCClient)
}
if cfg.AppConfig.MySQLDSN == "" || !strings.Contains(cfg.AppConfig.MySQLDSN, "hyapp_admin") {
t.Fatalf("gateway app config mysql dsn must point to admin config db: %+v", cfg.AppConfig)
}

View File

@ -15,8 +15,8 @@ import (
const (
serviceName = "gateway-service"
readyBudget = 500 * time.Millisecond
roomGRPCBudget = 300 * time.Millisecond
readyBudget = 2 * time.Second
grpcCheckBudget = 500 * time.Millisecond
componentStatus = "ok"
failStatus = "fail"
)
@ -28,22 +28,32 @@ type Check struct {
Message string `json:"message,omitempty"`
}
// GRPCDependency describes one internal service connection required before the
// gateway should receive new external traffic.
type GRPCDependency struct {
Name string
Conn *grpc.ClientConn
}
// State 记录 gateway 接新流量所需的进程状态和 room-service 连接状态。
type State struct {
nodeID string
jwtSecret string
room *grpc.ClientConn
nodeID string
jwtSecret string
dependencies []GRPCDependency
httpServing atomic.Bool
draining atomic.Bool
}
// NewState 创建 gateway 健康状态所有者。
func NewState(nodeID string, jwtSecret string, room *grpc.ClientConn) *State {
func NewState(nodeID string, jwtSecret string, room *grpc.ClientConn, dependencies ...GRPCDependency) *State {
all := make([]GRPCDependency, 0, 1+len(dependencies))
all = append(all, GRPCDependency{Name: "room_grpc", Conn: room})
all = append(all, dependencies...)
return &State{
nodeID: nodeID,
jwtSecret: jwtSecret,
room: room,
nodeID: nodeID,
jwtSecret: jwtSecret,
dependencies: all,
}
}
@ -90,13 +100,7 @@ func (s *State) Ready(ctx context.Context) []Check {
boolCheck("draining", !s.draining.Load(), "node is draining"),
}
roomCtx, roomCancel := context.WithTimeout(ctx, roomGRPCBudget)
defer roomCancel()
if err := waitGRPCReady(roomCtx, s.room); err != nil {
checks = append(checks, failCheck("room_grpc", err.Error()))
} else {
checks = append(checks, okCheck("room_grpc"))
}
checks = append(checks, s.grpcChecks(ctx)...)
return checks
}
@ -112,6 +116,36 @@ func Healthy(checks []Check) bool {
return true
}
func (s *State) grpcChecks(ctx context.Context) []Check {
results := make([]Check, len(s.dependencies))
done := make(chan struct{})
for index, dependency := range s.dependencies {
index := index
dependency := dependency
go func() {
name := strings.TrimSpace(dependency.Name)
if name == "" {
name = "grpc_dependency"
}
checkCtx, cancel := context.WithTimeout(ctx, grpcCheckBudget)
defer cancel()
if err := waitGRPCReady(checkCtx, dependency.Conn); err != nil {
results[index] = failCheck(name, err.Error())
} else {
results[index] = okCheck(name)
}
done <- struct{}{}
}()
}
for range s.dependencies {
<-done
}
return results
}
func waitGRPCReady(ctx context.Context, conn *grpc.ClientConn) error {
if conn == nil {
return errors.New("room gRPC connection is not initialized")

View File

@ -30,7 +30,7 @@ COPY services/room-service/configs/config.docker.yaml /app/config.yaml
USER appuser
EXPOSE 13001
EXPOSE 13001 13101
ENTRYPOINT ["/app/server"]
CMD ["-config", "/app/config.yaml"]

View File

@ -8,6 +8,11 @@ log:
include_response_body: false
max_payload_bytes: 2048
grpc_addr: ":13001"
advertise_addr: "room-service:13001"
health_http_addr: ":13101"
node_registry_ttl: "30s"
node_registry_heartbeat_interval: "10s"
owner_forward_timeout: "2s"
lease_ttl: "10s"
rank_limit: 20
snapshot_every_n: 10

View File

@ -8,6 +8,11 @@ log:
include_response_body: false
max_payload_bytes: 2048
grpc_addr: ":13001"
advertise_addr: "ROOM_NODE_PRIVATE_IP_OR_DNS:13001"
health_http_addr: ":13101"
node_registry_ttl: "30s"
node_registry_heartbeat_interval: "10s"
owner_forward_timeout: "2s"
lease_ttl: "10s"
rank_limit: 20
snapshot_every_n: 10

View File

@ -11,6 +11,11 @@ lease_ttl: "10s"
rank_limit: 20
snapshot_every_n: 10
grpc_addr: ":13001"
advertise_addr: "127.0.0.1:13001"
health_http_addr: ":13101"
node_registry_ttl: "30s"
node_registry_heartbeat_interval: "10s"
owner_forward_timeout: "2s"
wallet_service_addr: "127.0.0.1:13004"
activity_service_addr: "127.0.0.1:13006"
tencent_im:

View File

@ -4,6 +4,7 @@ package app
import (
"context"
"errors"
"log/slog"
"net"
"sync"
"time"
@ -14,6 +15,8 @@ import (
activityv1 "hyapp.local/api/proto/activity/v1"
roomv1 "hyapp.local/api/proto/room/v1"
"hyapp/pkg/grpchealth"
"hyapp/pkg/grpcshutdown"
"hyapp/pkg/healthhttp"
"hyapp/pkg/logx"
"hyapp/pkg/tencentim"
"hyapp/services/room-service/internal/config"
@ -43,8 +46,12 @@ type App struct {
repository *mysqlstorage.Repository
// redisClose 关闭 Redis clientRedis directory 本身不拥有额外生命周期。
redisClose func() error
// nodeRegistry 保存当前节点可被 owner 转发直连的实例地址。
nodeRegistry router.NodeRegistry
// health 汇总 gRPC、runtime、MySQL 和 Redis lease 探测状态。
health *healthcheck.State
// healthHTTP 给 CLB 和发布脚本提供 HTTP readiness不承载业务请求。
healthHTTP *healthhttp.Server
// workerCtx 统一控制本节点后台 worker关闭时必须先停止 worker 再释放 MySQL/Redis。
workerCtx context.Context
// workerCancel 停止 presence stale 和 outbox 补偿 worker。
@ -135,16 +142,26 @@ func New(cfg config.Config) (*App, error) {
}, directory, repository, integration.NewGRPCWalletClient(walletConn), roomPublisher, outboxPublisher)
server := grpc.NewServer(grpc.UnaryInterceptor(logx.UnaryServerInterceptor("room-service")))
roomServer := grpcserver.NewServer(svc, grpcserver.WithOwnerForwarding(cfg.NodeID, directory, directory, cfg.OwnerForwardTimeout))
// 命令服务处理房间状态变更,守卫服务给腾讯云 IM 回调或 gateway 做发言和进群校验。
roomv1.RegisterRoomCommandServiceServer(server, grpcserver.NewServer(svc))
roomv1.RegisterRoomGuardServiceServer(server, grpcserver.NewServer(svc))
roomv1.RegisterRoomQueryServiceServer(server, grpcserver.NewServer(svc))
roomv1.RegisterRoomCommandServiceServer(server, roomServer)
roomv1.RegisterRoomGuardServiceServer(server, roomServer)
roomv1.RegisterRoomQueryServiceServer(server, roomServer)
healthState := healthcheck.NewState(cfg.NodeID, svc, repository, directory)
healthgrpc.RegisterHealthServer(server, grpchealth.NewServer(healthState))
healthHTTP, err := healthhttp.New(cfg.HealthHTTPAddr, cfg.NodeID, healthState)
if err != nil {
_ = activityConn.Close()
_ = walletConn.Close()
_ = repository.Close()
_ = redisClient.Close()
return nil, err
}
// listener 在 New 阶段创建,可以把端口占用作为启动失败暴露。
listener, err := net.Listen("tcp", cfg.GRPCAddr)
if err != nil {
_ = healthHTTP.Close(context.Background())
_ = activityConn.Close()
_ = walletConn.Close()
_ = repository.Close()
@ -162,7 +179,9 @@ func New(cfg config.Config) (*App, error) {
activityConn: activityConn,
repository: repository,
redisClose: redisClient.Close,
nodeRegistry: directory,
health: healthState,
healthHTTP: healthHTTP,
workerCtx: workerCtx,
workerCancel: workerCancel,
}, nil
@ -170,14 +189,22 @@ func New(cfg config.Config) (*App, error) {
// Run 启动 gRPC 服务。
func (a *App) Run() error {
a.runHealthHTTP()
// health 先标记 serving再进入 Serveready 才能在监听开始后通过。
a.health.MarkGRPCServing()
if err := a.registerRoomNode(context.Background()); err != nil {
a.health.MarkGRPCStopped()
return err
}
defer func() {
if a.workerCancel != nil {
a.workerCancel()
}
a.workerWG.Wait()
}()
a.workerWG.Go(func() {
a.runNodeRegistryHeartbeat(a.workerCtx)
})
if a.cfg.PresenceStaleScanInterval > 0 {
// presence worker 只清理本节点已装载 Cell命令仍走 Room Cell 持久化链路。
a.workerWG.Go(func() {
@ -228,7 +255,15 @@ func (a *App) Close() {
}
a.workerWG.Wait()
// GracefulStop 让已进入的 gRPC 请求自然结束。
a.grpcServer.GracefulStop()
grpcshutdown.GracefulStop(a.grpcServer, 15*time.Second)
if a.service != nil {
// gRPC drain 完成后释放本节点持有的 room route降低滚动发布时等待 Redis TTL 的窗口。
releaseCtx, releaseCancel := context.WithTimeout(context.Background(), 5*time.Second)
a.service.ReleaseLoadedRoomLeases(releaseCtx)
releaseCancel()
}
a.unregisterRoomNode()
a.closeHealthHTTP()
if a.walletConn != nil {
_ = a.walletConn.Close()
}
@ -245,3 +280,74 @@ func (a *App) Close() {
}
})
}
func (a *App) registerRoomNode(ctx context.Context) error {
if a.nodeRegistry == nil {
return nil
}
if _, ok := ctx.Deadline(); !ok {
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, 2*time.Second)
defer cancel()
}
return a.nodeRegistry.RegisterNode(ctx, router.NodeRegistration{
NodeID: a.cfg.NodeID,
GRPCAddr: a.cfg.AdvertiseAddr,
}, a.cfg.NodeRegistryTTL)
}
func (a *App) runNodeRegistryHeartbeat(ctx context.Context) {
if a.nodeRegistry == nil {
return
}
ticker := time.NewTicker(a.cfg.NodeRegistryHeartbeatInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
if err := a.registerRoomNode(ctx); err != nil {
logx.Warn(ctx, "room_node_register_failed",
slog.String("node_id", a.cfg.NodeID),
slog.String("advertise_addr", a.cfg.AdvertiseAddr),
slog.String("error", err.Error()),
)
}
}
}
}
func (a *App) unregisterRoomNode() {
if a.nodeRegistry == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
if err := a.nodeRegistry.UnregisterNode(ctx, a.cfg.NodeID); err != nil {
logx.Warn(ctx, "room_node_unregister_failed",
slog.String("node_id", a.cfg.NodeID),
slog.String("error", err.Error()),
)
}
}
func (a *App) runHealthHTTP() {
if a.healthHTTP == nil {
return
}
go func() {
if err := a.healthHTTP.Run(); err != nil {
logx.Error(context.Background(), "health_http_run_failed", err)
}
}()
}
func (a *App) closeHealthHTTP() {
if a.healthHTTP == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
_ = a.healthHTTP.Close(ctx)
}

View File

@ -3,6 +3,7 @@ package config
import (
"fmt"
"net"
"net/http"
"strings"
"time"
@ -22,6 +23,16 @@ type Config struct {
NodeID string `yaml:"node_id"`
// GRPCAddr 是内部 gRPC 监听地址gateway-service 通过它访问 room-service。
GRPCAddr string `yaml:"grpc_addr"`
// AdvertiseAddr 是其他 room-service 实例转发到本节点时使用的私网地址,不能写负载均衡地址。
AdvertiseAddr string `yaml:"advertise_addr"`
// HealthHTTPAddr 是给 CLB/发布脚本探活使用的 HTTP 监听地址,不承载业务流量。
HealthHTTPAddr string `yaml:"health_http_addr"`
// NodeRegistryTTL 是 node_id -> advertise_addr 注册的有效期。
NodeRegistryTTL time.Duration `yaml:"node_registry_ttl"`
// NodeRegistryHeartbeatInterval 是刷新节点注册的周期。
NodeRegistryHeartbeatInterval time.Duration `yaml:"node_registry_heartbeat_interval"`
// OwnerForwardTimeout 是非 owner 节点转发到当前 owner 的单次 RPC 超时。
OwnerForwardTimeout time.Duration `yaml:"owner_forward_timeout"`
// LeaseTTL 是 room owner 租约有效期,节点故障后必须等它过期才能接管。
LeaseTTL time.Duration `yaml:"lease_ttl"`
// RankLimit 是房间本地礼物榜 top N。
@ -129,6 +140,10 @@ func Default() Config {
Environment: "local",
NodeID: "room-node-local",
GRPCAddr: ":13001",
AdvertiseAddr: "127.0.0.1:13001",
HealthHTTPAddr: ":13101",
NodeRegistryTTL: 30 * time.Second,
OwnerForwardTimeout: 2 * time.Second,
LeaseTTL: 10 * time.Second,
RankLimit: 20,
SnapshotEveryN: 10,
@ -190,6 +205,33 @@ func Normalize(cfg Config) (Config, error) {
if cfg.NodeID == "" {
cfg.NodeID = cfg.ServiceName
}
cfg.GRPCAddr = strings.TrimSpace(cfg.GRPCAddr)
if cfg.GRPCAddr == "" {
cfg.GRPCAddr = ":13001"
}
cfg.AdvertiseAddr = strings.TrimSpace(cfg.AdvertiseAddr)
if cfg.AdvertiseAddr == "" {
cfg.AdvertiseAddr = defaultAdvertiseAddr(cfg.GRPCAddr)
}
cfg.HealthHTTPAddr = strings.TrimSpace(cfg.HealthHTTPAddr)
if cfg.HealthHTTPAddr == "" {
cfg.HealthHTTPAddr = ":13101"
}
if cfg.NodeRegistryTTL <= 0 {
cfg.NodeRegistryTTL = 30 * time.Second
}
if cfg.NodeRegistryHeartbeatInterval <= 0 {
cfg.NodeRegistryHeartbeatInterval = cfg.NodeRegistryTTL / 3
}
if cfg.NodeRegistryHeartbeatInterval <= 0 {
cfg.NodeRegistryHeartbeatInterval = 10 * time.Second
}
if cfg.NodeRegistryHeartbeatInterval >= cfg.NodeRegistryTTL {
cfg.NodeRegistryHeartbeatInterval = cfg.NodeRegistryTTL / 3
}
if cfg.OwnerForwardTimeout <= 0 {
cfg.OwnerForwardTimeout = 2 * time.Second
}
cfg.WalletServiceAddr = strings.TrimSpace(cfg.WalletServiceAddr)
if cfg.WalletServiceAddr == "" {
cfg.WalletServiceAddr = "127.0.0.1:13004"
@ -207,6 +249,19 @@ func Normalize(cfg Config) (Config, error) {
return cfg, nil
}
func defaultAdvertiseAddr(grpcAddr string) string {
host, port, err := net.SplitHostPort(grpcAddr)
if err != nil {
// 非 host:port 形态保持原值,由启动拨测暴露错误。
return grpcAddr
}
if host == "" || host == "0.0.0.0" || host == "::" {
// 本地默认值只能服务单机开发;多机部署必须显式配置真实私网 IP 或主机名。
host = "127.0.0.1"
}
return net.JoinHostPort(host, port)
}
func normalizeOutboxWorkerConfig(cfg OutboxWorkerConfig) (OutboxWorkerConfig, error) {
// Outbox worker 的所有时间和批量参数都必须有安全下限,避免配置缺省导致 busy loop。
defaults := defaultOutboxWorkerConfig()

View File

@ -14,10 +14,16 @@ func TestLoad(t *testing.T) {
t.Fatalf("Load failed: %v", err)
}
// room-service 默认只暴露 gRPC,端口必须保持 13xxx 约束。
// room-service 业务只走 gRPCHTTP health 仅服务 CLB 探活,端口必须保持 13xxx 约束。
if cfg.GRPCAddr != ":13001" {
t.Fatalf("unexpected grpc addr: %s", cfg.GRPCAddr)
}
if cfg.HealthHTTPAddr != ":13101" {
t.Fatalf("unexpected health http addr: %s", cfg.HealthHTTPAddr)
}
if cfg.AdvertiseAddr != "127.0.0.1:13001" || cfg.NodeRegistryTTL != 30*time.Second || cfg.NodeRegistryHeartbeatInterval != 10*time.Second || cfg.OwnerForwardTimeout != 2*time.Second {
t.Fatalf("unexpected owner routing config: advertise=%s ttl=%s heartbeat=%s timeout=%s", cfg.AdvertiseAddr, cfg.NodeRegistryTTL, cfg.NodeRegistryHeartbeatInterval, cfg.OwnerForwardTimeout)
}
// lease TTL 是故障接管抖动窗口的基础参数。
if cfg.LeaseTTL != 10*time.Second {
@ -55,6 +61,9 @@ func TestLoadTencentExample(t *testing.T) {
if cfg.TencentIM.RequestTimeout != 5*time.Second {
t.Fatalf("unexpected request timeout: %s", cfg.TencentIM.RequestTimeout)
}
if cfg.AdvertiseAddr == "" || cfg.AdvertiseAddr == "room-service.internal:13001" {
t.Fatalf("tencent example must use per-instance advertise_addr, got %q", cfg.AdvertiseAddr)
}
if !cfg.OutboxWorker.Enabled || cfg.OutboxWorker.RetryStrategy != "exponential_backoff" || cfg.OutboxWorker.MaxRetryCount != 10 {
t.Fatalf("tencent example must configure outbox worker: %+v", cfg.OutboxWorker)
}

View File

@ -53,6 +53,12 @@ func runtimeRoomKey(appCode string, roomID string) string {
return appCode + "\x00" + roomID
}
// RuntimeRoomKey 返回 Redis owner lease 使用的运行时路由 key。
// transport 层 owner 转发必须和领域层 EnsureOwner 使用同一个 key否则会出现同一房间多 owner。
func RuntimeRoomKey(appCode string, roomID string) string {
return runtimeRoomKey(appCode, roomID)
}
func runtimeRoomKeyFromContext(ctx context.Context, roomID string) string {
return runtimeRoomKey(appcode.FromContext(ctx), roomID)
}

View File

@ -0,0 +1,35 @@
package service
import (
"context"
"log/slog"
"hyapp/pkg/logx"
)
// ReleaseLoadedRoomLeases 释放当前节点仍持有的已加载房间 lease。
// 该方法只应在 gRPC 已停止接新请求并完成 in-flight drain 后调用;提前释放会破坏单房间单 owner。
func (s *Service) ReleaseLoadedRoomLeases(ctx context.Context) {
if s == nil || s.directory == nil {
return
}
for _, roomRef := range s.loadedRoomRefs() {
released, err := s.directory.ReleaseOwner(ctx, runtimeRoomKey(roomRef.AppCode, roomRef.RoomID), s.nodeID)
if err != nil {
logx.Warn(ctx, "room_lease_release_failed",
slog.String("node_id", s.nodeID),
slog.String("app_code", roomRef.AppCode),
slog.String("room_id", roomRef.RoomID),
slog.String("error", err.Error()),
)
continue
}
if released {
logx.Info(ctx, "room_lease_released",
slog.String("node_id", s.nodeID),
slog.String("app_code", roomRef.AppCode),
slog.String("room_id", roomRef.RoomID),
)
}
}
}

File diff suppressed because it is too large Load Diff

View File

@ -21,6 +21,24 @@ type Lease struct {
ExpiresAt time.Time
}
// NodeRegistration 记录一个 room-service 实例可被其他实例直连的内网地址。
type NodeRegistration struct {
// NodeID 必须和 Redis lease 中的 owner node_id 完全一致。
NodeID string
// GRPCAddr 是实例私网 gRPC 地址,必须指向单个实例,不能写 room-service 负载均衡地址。
GRPCAddr string
// UpdatedAt 是最近一次心跳写入时间,便于排查节点注册是否卡住。
UpdatedAt time.Time
// ExpiresAt 是节点注册过期时间;过期节点不能作为 owner 转发目标。
ExpiresAt time.Time
}
// ValidAt 判断节点注册在指定时间是否仍可作为转发目标。
func (n NodeRegistration) ValidAt(now time.Time) bool {
// 没有地址或过期时间的注册不能被使用,避免把请求转发到空地址或旧实例。
return n.NodeID != "" && n.GRPCAddr != "" && !n.ExpiresAt.IsZero() && now.Before(n.ExpiresAt)
}
// ValidAt 判断租约在指定时刻是否仍有效。
func (l Lease) ValidAt(now time.Time) bool {
// 零值过期时间永远无效,避免空 lease 被误认为可继续持有。
@ -35,22 +53,37 @@ type Directory interface {
EnsureOwner(ctx context.Context, roomID string, nodeID string, now time.Time, ttl time.Duration) (Lease, error)
// VerifyOwner 校验当前节点和 lease token 仍然持有执行权,不续租也不接管。
VerifyOwner(ctx context.Context, roomID string, nodeID string, leaseToken string, now time.Time) (bool, error)
// ReleaseOwner 在节点安全下线后释放自己仍持有的 owner lease让新节点无需等待 TTL 才能恢复。
ReleaseOwner(ctx context.Context, roomID string, nodeID string) (bool, error)
// ForceExpire 只用于测试模拟租约过期。
ForceExpire(roomID string)
}
// NodeRegistry 维护 room-service node_id 到实例私网 gRPC 地址的短租约。
type NodeRegistry interface {
// RegisterNode 写入或续约当前节点地址ttl 过期后其他节点不能继续转发到该实例。
RegisterNode(ctx context.Context, node NodeRegistration, ttl time.Duration) error
// LookupNode 查询一个 owner node_id 的实例地址。
LookupNode(ctx context.Context, nodeID string) (NodeRegistration, bool, error)
// UnregisterNode 在实例安全退出后删除注册,避免转发到已停止进程。
UnregisterNode(ctx context.Context, nodeID string) error
}
// MemoryDirectory 是首版测试与本地运行使用的内存目录实现。
type MemoryDirectory struct {
// mu 保护 leases测试里可能用多个服务实例模拟接管。
mu sync.Mutex
// leases 保存 room_id -> lease。
leases map[string]Lease
// nodes 保存 node_id -> 注册地址,供 transport 层 owner 转发测试复用。
nodes map[string]NodeRegistration
}
// NewMemoryDirectory 初始化空的内存目录。
func NewMemoryDirectory() *MemoryDirectory {
return &MemoryDirectory{
leases: make(map[string]Lease),
nodes: make(map[string]NodeRegistration),
}
}
@ -112,6 +145,20 @@ func (d *MemoryDirectory) VerifyOwner(_ context.Context, roomID string, nodeID s
return current.NodeID == nodeID && current.LeaseToken == leaseToken && current.ValidAt(now), nil
}
// ReleaseOwner 删除当前节点持有的内存 lease其他节点持有时保持不变。
func (d *MemoryDirectory) ReleaseOwner(_ context.Context, roomID string, nodeID string) (bool, error) {
d.mu.Lock()
defer d.mu.Unlock()
current, ok := d.leases[roomID]
if !ok || current.NodeID != nodeID {
return false, nil
}
delete(d.leases, roomID)
return true, nil
}
// ForceExpire 只用于测试时模拟租约自然过期。
func (d *MemoryDirectory) ForceExpire(roomID string) {
d.mu.Lock()
@ -126,3 +173,40 @@ func (d *MemoryDirectory) ForceExpire(roomID string) {
lease.ExpiresAt = time.Unix(0, 0)
d.leases[roomID] = lease
}
// RegisterNode 写入测试用节点地址。
func (d *MemoryDirectory) RegisterNode(_ context.Context, node NodeRegistration, ttl time.Duration) error {
d.mu.Lock()
defer d.mu.Unlock()
now := time.Now().UTC()
if ttl <= 0 {
ttl = 30 * time.Second
}
node.UpdatedAt = now
node.ExpiresAt = now.Add(ttl)
d.nodes[node.NodeID] = node
return nil
}
// LookupNode 查询测试用节点地址,过期注册按不存在处理。
func (d *MemoryDirectory) LookupNode(_ context.Context, nodeID string) (NodeRegistration, bool, error) {
d.mu.Lock()
defer d.mu.Unlock()
node, ok := d.nodes[nodeID]
if !ok || !node.ValidAt(time.Now().UTC()) {
return NodeRegistration{}, false, nil
}
return node, true, nil
}
// UnregisterNode 删除测试用节点地址。
func (d *MemoryDirectory) UnregisterNode(_ context.Context, nodeID string) error {
d.mu.Lock()
defer d.mu.Unlock()
delete(d.nodes, nodeID)
return nil
}

View File

@ -4,7 +4,9 @@ package router
import (
"context"
"encoding/json"
"fmt"
"strconv"
"strings"
"time"
"github.com/redis/go-redis/v9"
@ -14,6 +16,9 @@ import (
// roomRouteKeyPrefix 是生产房间 owner lease 的 Redis key 前缀。
const roomRouteKeyPrefix = "room:route:"
// roomNodeKeyPrefix 保存 room-service 节点的实例直连地址owner 转发只读取该短租约。
const roomNodeKeyPrefix = "room:node:"
// healthLeaseKey 是 ready 探针专用 key避免探针污染真实房间路由。
const healthLeaseKey = "room:healthcheck:lease"
@ -70,6 +75,25 @@ end
return 0
`
const releaseOwnerScript = `
local current = redis.call("GET", KEYS[1])
if not current then
return 0
end
local ok, decoded = pcall(cjson.decode, current)
if not ok then
return 0
end
if decoded["node_id"] == ARGV[1] then
redis.call("DEL", KEYS[1])
return 1
end
return 0
`
// RedisDirectory 用 Redis 保存 room_id 到执行节点的带 TTL 租约。
type RedisDirectory struct {
// client 是 go-redis 客户端,连接生命周期由 app.App 关闭。
@ -84,6 +108,14 @@ type redisLeaseValue struct {
ExpiresAt int64 `json:"expires_at_ms"`
}
// redisNodeValue 是 Redis 中的节点注册 JSON字段保持可读以便 redis-cli 排查。
type redisNodeValue struct {
NodeID string `json:"node_id"`
GRPCAddr string `json:"grpc_addr"`
UpdatedAt int64 `json:"updated_at_ms"`
ExpiresAt int64 `json:"expires_at_ms"`
}
// NewRedisDirectory 创建 Redis 目录实现。
func NewRedisDirectory(client *redis.Client) *RedisDirectory {
return &RedisDirectory{client: client}
@ -222,17 +254,80 @@ func (d *RedisDirectory) VerifyOwner(ctx context.Context, roomID string, nodeID
return value == 1, nil
}
// ReleaseOwner 在节点安全下线后删除自己持有的 route lease。
func (d *RedisDirectory) ReleaseOwner(ctx context.Context, roomID string, nodeID string) (bool, error) {
value, err := d.client.Eval(ctx, releaseOwnerScript, []string{routeKey(roomID)}, nodeID).Int()
if err != nil {
return false, err
}
return value == 1, nil
}
// ForceExpire 只服务测试,生产启动路径不会调用。
func (d *RedisDirectory) ForceExpire(roomID string) {
// Redis 实现直接删除 key 来模拟 lease 过期,避免等待真实 TTL。
_ = d.client.Del(context.Background(), routeKey(roomID)).Err()
}
// RegisterNode 写入当前 room-service 实例地址TTL 过期后 owner 转发不会再使用该节点。
func (d *RedisDirectory) RegisterNode(ctx context.Context, node NodeRegistration, ttl time.Duration) error {
if ttl <= 0 {
ttl = 30 * time.Second
}
node.NodeID = strings.TrimSpace(node.NodeID)
node.GRPCAddr = strings.TrimSpace(node.GRPCAddr)
if node.NodeID == "" || node.GRPCAddr == "" {
return fmt.Errorf("node_id and grpc_addr are required")
}
now := time.Now().UTC()
node.UpdatedAt = now
node.ExpiresAt = now.Add(ttl)
payload, err := encodeNode(node)
if err != nil {
return err
}
return d.client.Set(ctx, nodeKey(node.NodeID), payload, ttl).Err()
}
// LookupNode 查询 owner node_id 对应的实例地址。
func (d *RedisDirectory) LookupNode(ctx context.Context, nodeID string) (NodeRegistration, bool, error) {
value, err := d.client.Get(ctx, nodeKey(nodeID)).Result()
if err != nil {
if err == redis.Nil {
return NodeRegistration{}, false, nil
}
return NodeRegistration{}, false, err
}
node, err := decodeNode(value)
if err != nil {
return NodeRegistration{}, false, err
}
if !node.ValidAt(time.Now().UTC()) {
return NodeRegistration{}, false, nil
}
return node, true, nil
}
// UnregisterNode 删除节点注册;下线节点停止接受 owner 转发前必须调用。
func (d *RedisDirectory) UnregisterNode(ctx context.Context, nodeID string) error {
return d.client.Del(ctx, nodeKey(nodeID)).Err()
}
func routeKey(roomID string) string {
// 路由 key 只由房间 ID 决定,所有节点必须计算出同一个 Redis key。
return roomRouteKeyPrefix + roomID
}
func nodeKey(nodeID string) string {
// 节点 key 使用 node_id不包含 app_code一个实例可以服务多个 App 的房间。
return roomNodeKeyPrefix + strings.TrimSpace(nodeID)
}
func encodeLease(lease Lease) (string, error) {
// 使用 JSON 而不是二进制,方便线上通过 redis-cli 直接排查 owner。
payload, err := json.Marshal(redisLeaseValue{
@ -248,6 +343,20 @@ func encodeLease(lease Lease) (string, error) {
return string(payload), nil
}
func encodeNode(node NodeRegistration) (string, error) {
payload, err := json.Marshal(redisNodeValue{
NodeID: strings.TrimSpace(node.NodeID),
GRPCAddr: strings.TrimSpace(node.GRPCAddr),
UpdatedAt: node.UpdatedAt.UnixMilli(),
ExpiresAt: node.ExpiresAt.UnixMilli(),
})
if err != nil {
return "", err
}
return string(payload), nil
}
func decodeLease(value string) (Lease, error) {
var payload redisLeaseValue
if err := json.Unmarshal([]byte(value), &payload); err != nil {
@ -263,3 +372,17 @@ func decodeLease(value string) (Lease, error) {
ExpiresAt: time.UnixMilli(payload.ExpiresAt),
}, nil
}
func decodeNode(value string) (NodeRegistration, error) {
var payload redisNodeValue
if err := json.Unmarshal([]byte(value), &payload); err != nil {
return NodeRegistration{}, err
}
return NodeRegistration{
NodeID: strings.TrimSpace(payload.NodeID),
GRPCAddr: strings.TrimSpace(payload.GRPCAddr),
UpdatedAt: time.UnixMilli(payload.UpdatedAt),
ExpiresAt: time.UnixMilli(payload.ExpiresAt),
}, nil
}

View File

@ -0,0 +1,172 @@
package grpc
import (
"context"
"fmt"
"strconv"
"strings"
"sync"
"time"
"hyapp/pkg/xerr"
roomservice "hyapp/services/room-service/internal/room/service"
"hyapp/services/room-service/internal/router"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/metadata"
roomv1 "hyapp.local/api/proto/room/v1"
)
const (
// ownerForwardDepthHeader 防止错误注册到负载均衡地址后出现递归转发。
ownerForwardDepthHeader = "x-hyapp-room-owner-forward-depth"
// maxOwnerForwardDepth 允许一次中转;超过后让本节点走本地逻辑并返回真实 owner 冲突。
maxOwnerForwardDepth = 1
// defaultOwnerForwardTimeout 限制转发占用调用方请求预算的最长时间。
defaultOwnerForwardTimeout = 2 * time.Second
)
type ownerForwarder struct {
// nodeID 是当前 room-service 实例 ID用于判断自己是否就是 owner。
nodeID string
// directory 保存 room_id -> owner node_id 的短租约。
directory router.Directory
// registry 保存 owner node_id -> 实例 gRPC 地址的短租约。
registry router.NodeRegistry
// timeout 是单次 owner 转发的额外 deadline 上限。
timeout time.Duration
mu sync.Mutex
conns map[string]forwardConn
}
type forwardConn struct {
addr string
conn *grpc.ClientConn
}
func newOwnerForwarder(nodeID string, directory router.Directory, registry router.NodeRegistry, timeout time.Duration) *ownerForwarder {
if timeout <= 0 {
timeout = defaultOwnerForwardTimeout
}
return &ownerForwarder{
nodeID: strings.TrimSpace(nodeID),
directory: directory,
registry: registry,
timeout: timeout,
conns: make(map[string]forwardConn),
}
}
func forwardCommandToOwner[T any](f *ownerForwarder, ctx context.Context, appCode string, roomID string, call func(context.Context, roomv1.RoomCommandServiceClient) (T, error)) (T, bool, error) {
return forwardToOwner(ctx, f, appCode, roomID, func(callCtx context.Context, conn *grpc.ClientConn) (T, error) {
return call(callCtx, roomv1.NewRoomCommandServiceClient(conn))
})
}
func forwardGuardToOwner[T any](f *ownerForwarder, ctx context.Context, appCode string, roomID string, call func(context.Context, roomv1.RoomGuardServiceClient) (T, error)) (T, bool, error) {
return forwardToOwner(ctx, f, appCode, roomID, func(callCtx context.Context, conn *grpc.ClientConn) (T, error) {
return call(callCtx, roomv1.NewRoomGuardServiceClient(conn))
})
}
func forwardQueryToOwner[T any](f *ownerForwarder, ctx context.Context, appCode string, roomID string, call func(context.Context, roomv1.RoomQueryServiceClient) (T, error)) (T, bool, error) {
return forwardToOwner(ctx, f, appCode, roomID, func(callCtx context.Context, conn *grpc.ClientConn) (T, error) {
return call(callCtx, roomv1.NewRoomQueryServiceClient(conn))
})
}
func forwardToOwner[T any](ctx context.Context, f *ownerForwarder, appCode string, roomID string, call func(context.Context, *grpc.ClientConn) (T, error)) (T, bool, error) {
var zero T
if f == nil || f.directory == nil || f.registry == nil || strings.TrimSpace(roomID) == "" {
return zero, false, nil
}
if ownerForwardDepth(ctx) >= maxOwnerForwardDepth {
// 已经中转过的请求继续走本地逻辑,避免注册错误造成递归 RPC。
return zero, false, nil
}
now := time.Now().UTC()
lease, exists, err := f.directory.Lookup(ctx, roomservice.RuntimeRoomKey(appCode, roomID))
if err != nil {
return zero, true, xerr.ToGRPCError(xerr.New(xerr.Unavailable, "room owner lookup failed"))
}
if !exists || !lease.ValidAt(now) || lease.NodeID == f.nodeID {
return zero, false, nil
}
node, exists, err := f.registry.LookupNode(ctx, lease.NodeID)
if err != nil {
return zero, true, xerr.ToGRPCError(xerr.New(xerr.Unavailable, "room owner node lookup failed"))
}
if !exists || !node.ValidAt(now) {
return zero, true, xerr.ToGRPCError(xerr.New(xerr.Unavailable, "room owner node is not registered"))
}
conn, err := f.connForNode(ctx, node)
if err != nil {
return zero, true, xerr.ToGRPCError(xerr.New(xerr.Unavailable, fmt.Sprintf("room owner node dial failed: %v", err)))
}
callCtx, cancel := context.WithTimeout(ctx, f.timeout)
defer cancel()
callCtx = metadata.AppendToOutgoingContext(callCtx, ownerForwardDepthHeader, strconv.Itoa(ownerForwardDepth(ctx)+1))
result, err := call(callCtx, conn)
return result, true, err
}
func (f *ownerForwarder) connForNode(_ context.Context, node router.NodeRegistration) (*grpc.ClientConn, error) {
f.mu.Lock()
defer f.mu.Unlock()
if cached, ok := f.conns[node.NodeID]; ok && cached.addr == node.GRPCAddr {
return cached.conn, nil
}
if cached, ok := f.conns[node.NodeID]; ok && cached.conn != nil {
// 节点地址变化通常来自实例重建;关闭旧连接,避免继续向旧进程转发。
_ = cached.conn.Close()
}
conn, err := grpc.Dial(node.GRPCAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
return nil, err
}
f.conns[node.NodeID] = forwardConn{addr: node.GRPCAddr, conn: conn}
return conn, nil
}
func (f *ownerForwarder) close() {
if f == nil {
return
}
f.mu.Lock()
defer f.mu.Unlock()
for nodeID, cached := range f.conns {
if cached.conn != nil {
_ = cached.conn.Close()
}
delete(f.conns, nodeID)
}
}
func ownerForwardDepth(ctx context.Context) int {
var values []string
if incoming, ok := metadata.FromIncomingContext(ctx); ok {
values = incoming.Get(ownerForwardDepthHeader)
}
if len(values) == 0 {
if outgoing, ok := metadata.FromOutgoingContext(ctx); ok {
values = outgoing.Get(ownerForwardDepthHeader)
}
}
if len(values) == 0 {
return 0
}
depth, err := strconv.Atoi(strings.TrimSpace(values[len(values)-1]))
if err != nil || depth < 0 {
return 0
}
return depth
}

View File

@ -3,11 +3,13 @@ package grpc
import (
"context"
"time"
roomv1 "hyapp.local/api/proto/room/v1"
"hyapp/pkg/appcode"
"hyapp/pkg/xerr"
roomservice "hyapp/services/room-service/internal/room/service"
"hyapp/services/room-service/internal/router"
)
// Server 把 room-service 领域实现挂到 gRPC 接口上。
@ -21,11 +23,29 @@ type Server struct {
// svc 是领域服务入口gRPC 层不保存任何房间状态。
svc *roomservice.Service
// ownerForwarder 在请求命中非 owner 节点时把房间命令转到当前 owner 实例。
ownerForwarder *ownerForwarder
}
// Option 调整 room-service gRPC 边界行为。
type Option func(*Server)
// WithOwnerForwarding 启用 room_id -> owner node -> instance addr 转发。
func WithOwnerForwarding(nodeID string, directory router.Directory, registry router.NodeRegistry, timeout time.Duration) Option {
return func(s *Server) {
s.ownerForwarder = newOwnerForwarder(nodeID, directory, registry, timeout)
}
}
// NewServer 创建一个可注册到 gRPC 的 room server。
func NewServer(svc *roomservice.Service) *Server {
return &Server{svc: svc}
func NewServer(svc *roomservice.Service, opts ...Option) *Server {
server := &Server{svc: svc}
for _, opt := range opts {
if opt != nil {
opt(server)
}
}
return server
}
// mapServiceResult 把领域错误固定转换为 gRPC status + ErrorInfo.reason。
@ -44,6 +64,11 @@ func mapServiceResult[T any](result T, err error) (T, error) {
func (s *Server) CreateRoom(ctx context.Context, req *roomv1.CreateRoomRequest) (*roomv1.CreateRoomResponse, error) {
// 创建房间会获取 Redis lease、写 MySQL meta/command/outbox/snapshot并安装本地 Cell。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.CreateRoomResponse, error) {
return client.CreateRoom(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.CreateRoom(ctx, req))
}
@ -51,6 +76,11 @@ func (s *Server) CreateRoom(ctx context.Context, req *roomv1.CreateRoomRequest)
func (s *Server) UpdateRoomProfile(ctx context.Context, req *roomv1.UpdateRoomProfileRequest) (*roomv1.UpdateRoomProfileResponse, error) {
// 房间资料和座位数变更必须经 Room Cell 串行提交gRPC 层不直接修改快照。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.UpdateRoomProfileResponse, error) {
return client.UpdateRoomProfile(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.UpdateRoomProfile(ctx, req))
}
@ -58,6 +88,11 @@ func (s *Server) UpdateRoomProfile(ctx context.Context, req *roomv1.UpdateRoomPr
func (s *Server) JoinRoom(ctx context.Context, req *roomv1.JoinRoomRequest) (*roomv1.JoinRoomResponse, error) {
// JoinRoom 只维护 room-service presence真实消息连接由腾讯云 IM SDK 处理。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.JoinRoomResponse, error) {
return client.JoinRoom(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.JoinRoom(ctx, req))
}
@ -65,6 +100,11 @@ func (s *Server) JoinRoom(ctx context.Context, req *roomv1.JoinRoomRequest) (*ro
func (s *Server) RoomHeartbeat(ctx context.Context, req *roomv1.RoomHeartbeatRequest) (*roomv1.RoomHeartbeatResponse, error) {
// 心跳只刷新已有业务 presence不能隐式创建进房状态。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.RoomHeartbeatResponse, error) {
return client.RoomHeartbeat(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.RoomHeartbeat(ctx, req))
}
@ -72,6 +112,11 @@ func (s *Server) RoomHeartbeat(ctx context.Context, req *roomv1.RoomHeartbeatReq
func (s *Server) LeaveRoom(ctx context.Context, req *roomv1.LeaveRoomRequest) (*roomv1.LeaveRoomResponse, error) {
// LeaveRoom 不直接操作腾讯云 IM 连接,只修改房间业务态。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.LeaveRoomResponse, error) {
return client.LeaveRoom(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.LeaveRoom(ctx, req))
}
@ -79,6 +124,11 @@ func (s *Server) LeaveRoom(ctx context.Context, req *roomv1.LeaveRoomRequest) (*
func (s *Server) CloseRoom(ctx context.Context, req *roomv1.CloseRoomRequest) (*roomv1.CloseRoomResponse, error) {
// CloseRoom 统一关闭房间生命周期、清理 presence/麦位并写出系统事件。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.CloseRoomResponse, error) {
return client.CloseRoom(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.CloseRoom(ctx, req))
}
@ -86,6 +136,11 @@ func (s *Server) CloseRoom(ctx context.Context, req *roomv1.CloseRoomRequest) (*
func (s *Server) MicUp(ctx context.Context, req *roomv1.MicUpRequest) (*roomv1.MicUpResponse, error) {
// 麦位修改由 Room Cell 串行执行gRPC 层不做状态判断。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.MicUpResponse, error) {
return client.MicUp(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.MicUp(ctx, req))
}
@ -93,6 +148,11 @@ func (s *Server) MicUp(ctx context.Context, req *roomv1.MicUpRequest) (*roomv1.M
func (s *Server) MicDown(ctx context.Context, req *roomv1.MicDownRequest) (*roomv1.MicDownResponse, error) {
// 下麦结果会进入 command log 和 outbox失败不产生部分状态。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.MicDownResponse, error) {
return client.MicDown(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.MicDown(ctx, req))
}
@ -100,6 +160,11 @@ func (s *Server) MicDown(ctx context.Context, req *roomv1.MicDownRequest) (*room
func (s *Server) ChangeMicSeat(ctx context.Context, req *roomv1.ChangeMicSeatRequest) (*roomv1.ChangeMicSeatResponse, error) {
// 换位在领域层原子修改源麦位和目标麦位。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.ChangeMicSeatResponse, error) {
return client.ChangeMicSeat(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.ChangeMicSeat(ctx, req))
}
@ -107,6 +172,11 @@ func (s *Server) ChangeMicSeat(ctx context.Context, req *roomv1.ChangeMicSeatReq
func (s *Server) ConfirmMicPublishing(ctx context.Context, req *roomv1.ConfirmMicPublishingRequest) (*roomv1.ConfirmMicPublishingResponse, error) {
// 发流确认必须进入 Room Cell 串行状态机,避免旧 RTC 事件覆盖新 mic_session。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.ConfirmMicPublishingResponse, error) {
return client.ConfirmMicPublishing(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.ConfirmMicPublishing(ctx, req))
}
@ -114,6 +184,11 @@ func (s *Server) ConfirmMicPublishing(ctx context.Context, req *roomv1.ConfirmMi
func (s *Server) SetMicMute(ctx context.Context, req *roomv1.SetMicMuteRequest) (*roomv1.SetMicMuteResponse, error) {
// 服务端可见静音态必须经 Room Cell 提交,确保其他用户 UI 和系统消息一致。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.SetMicMuteResponse, error) {
return client.SetMicMute(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.SetMicMute(ctx, req))
}
@ -121,6 +196,11 @@ func (s *Server) SetMicMute(ctx context.Context, req *roomv1.SetMicMuteRequest)
func (s *Server) ApplyRTCEvent(ctx context.Context, req *roomv1.ApplyRTCEventRequest) (*roomv1.ApplyRTCEventResponse, error) {
// 腾讯 RTC 回调已在 gateway 验签;房间状态变化仍必须由 Room Cell 串行提交。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.ApplyRTCEventResponse, error) {
return client.ApplyRTCEvent(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.ApplyRTCEvent(ctx, req))
}
@ -128,6 +208,11 @@ func (s *Server) ApplyRTCEvent(ctx context.Context, req *roomv1.ApplyRTCEventReq
func (s *Server) SetMicSeatLock(ctx context.Context, req *roomv1.SetMicSeatLockRequest) (*roomv1.SetMicSeatLockResponse, error) {
// 锁麦权限和目标麦位状态由 Room Cell 判定。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.SetMicSeatLockResponse, error) {
return client.SetMicSeatLock(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.SetMicSeatLock(ctx, req))
}
@ -135,6 +220,11 @@ func (s *Server) SetMicSeatLock(ctx context.Context, req *roomv1.SetMicSeatLockR
func (s *Server) SetChatEnabled(ctx context.Context, req *roomv1.SetChatEnabledRequest) (*roomv1.SetChatEnabledResponse, error) {
// 公屏开关是 room-service 发言守卫的权威状态。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.SetChatEnabledResponse, error) {
return client.SetChatEnabled(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.SetChatEnabled(ctx, req))
}
@ -142,6 +232,11 @@ func (s *Server) SetChatEnabled(ctx context.Context, req *roomv1.SetChatEnabledR
func (s *Server) SetRoomAdmin(ctx context.Context, req *roomv1.SetRoomAdminRequest) (*roomv1.SetRoomAdminResponse, error) {
// 管理员集合只能由 room-service 权限矩阵修改。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.SetRoomAdminResponse, error) {
return client.SetRoomAdmin(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.SetRoomAdmin(ctx, req))
}
@ -149,6 +244,11 @@ func (s *Server) SetRoomAdmin(ctx context.Context, req *roomv1.SetRoomAdminReque
func (s *Server) TransferRoomHost(ctx context.Context, req *roomv1.TransferRoomHostRequest) (*roomv1.TransferRoomHostResponse, error) {
// 主持人转移由 Room Cell 串行提交,并进入 command log/outbox。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.TransferRoomHostResponse, error) {
return client.TransferRoomHost(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.TransferRoomHost(ctx, req))
}
@ -156,6 +256,11 @@ func (s *Server) TransferRoomHost(ctx context.Context, req *roomv1.TransferRoomH
func (s *Server) MuteUser(ctx context.Context, req *roomv1.MuteUserRequest) (*roomv1.MuteUserResponse, error) {
// 禁言状态是腾讯云 IM 发言回调的权威来源。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.MuteUserResponse, error) {
return client.MuteUser(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.MuteUser(ctx, req))
}
@ -163,6 +268,11 @@ func (s *Server) MuteUser(ctx context.Context, req *roomv1.MuteUserRequest) (*ro
func (s *Server) KickUser(ctx context.Context, req *roomv1.KickUserRequest) (*roomv1.KickUserResponse, error) {
// 踢人会修改 presence、ban 集合和麦位占用。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.KickUserResponse, error) {
return client.KickUser(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.KickUser(ctx, req))
}
@ -170,6 +280,11 @@ func (s *Server) KickUser(ctx context.Context, req *roomv1.KickUserRequest) (*ro
func (s *Server) UnbanUser(ctx context.Context, req *roomv1.UnbanUserRequest) (*roomv1.UnbanUserResponse, error) {
// 解封只解除 ban不恢复 presence、管理员身份或麦位。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.UnbanUserResponse, error) {
return client.UnbanUser(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.UnbanUser(ctx, req))
}
@ -177,6 +292,11 @@ func (s *Server) UnbanUser(ctx context.Context, req *roomv1.UnbanUserRequest) (*
func (s *Server) SendGift(ctx context.Context, req *roomv1.SendGiftRequest) (*roomv1.SendGiftResponse, error) {
// SendGift 先同步扣费,成功后才更新房间热度和本地榜。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardCommand(s, ctx, req.GetMeta(), func(callCtx context.Context, client roomv1.RoomCommandServiceClient) (*roomv1.SendGiftResponse, error) {
return client.SendGift(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.SendGift(ctx, req))
}
@ -184,6 +304,11 @@ func (s *Server) SendGift(ctx context.Context, req *roomv1.SendGiftRequest) (*ro
func (s *Server) CheckSpeakPermission(ctx context.Context, req *roomv1.CheckSpeakPermissionRequest) (*roomv1.CheckSpeakPermissionResponse, error) {
// 外部 IM 发公屏前调用该守卫,避免复制禁言和房间状态规则。
ctx = appcode.WithContext(ctx, req.GetAppCode())
if resp, forwarded, err := forwardGuard(s, ctx, req.GetAppCode(), req.GetRoomId(), func(callCtx context.Context, client roomv1.RoomGuardServiceClient) (*roomv1.CheckSpeakPermissionResponse, error) {
return client.CheckSpeakPermission(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.CheckSpeakPermission(ctx, req))
}
@ -191,6 +316,11 @@ func (s *Server) CheckSpeakPermission(ctx context.Context, req *roomv1.CheckSpea
func (s *Server) VerifyRoomPresence(ctx context.Context, req *roomv1.VerifyRoomPresenceRequest) (*roomv1.VerifyRoomPresenceResponse, error) {
// 外部 IM 进群前调用该守卫,防止未进房或被踢用户收消息。
ctx = appcode.WithContext(ctx, req.GetAppCode())
if resp, forwarded, err := forwardGuard(s, ctx, req.GetAppCode(), req.GetRoomId(), func(callCtx context.Context, client roomv1.RoomGuardServiceClient) (*roomv1.VerifyRoomPresenceResponse, error) {
return client.VerifyRoomPresence(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.VerifyRoomPresence(ctx, req))
}
@ -226,6 +356,11 @@ func (s *Server) GetCurrentRoom(ctx context.Context, req *roomv1.GetCurrentRoomR
func (s *Server) GetRoomSnapshot(ctx context.Context, req *roomv1.GetRoomSnapshotRequest) (*roomv1.GetRoomSnapshotResponse, error) {
// 快照查询不刷新 presence不替代 heartbeat也不能让未进房用户读取完整房间态。
ctx = contextWithMetaApp(ctx, req.GetMeta())
if resp, forwarded, err := forwardQuery(s, ctx, req.GetMeta().GetAppCode(), req.GetRoomId(), func(callCtx context.Context, client roomv1.RoomQueryServiceClient) (*roomv1.GetRoomSnapshotResponse, error) {
return client.GetRoomSnapshot(callCtx, req)
}); forwarded {
return resp, err
}
return mapServiceResult(s.svc.GetRoomSnapshot(ctx, req))
}
@ -242,3 +377,27 @@ func contextWithMetaApp(ctx context.Context, meta *roomv1.RequestMeta) context.C
}
return appcode.WithContext(ctx, meta.GetAppCode())
}
func forwardCommand[T any](s *Server, ctx context.Context, meta *roomv1.RequestMeta, call func(context.Context, roomv1.RoomCommandServiceClient) (T, error)) (T, bool, error) {
var zero T
if s == nil || s.ownerForwarder == nil || meta == nil {
return zero, false, nil
}
return forwardCommandToOwner(s.ownerForwarder, ctx, meta.GetAppCode(), meta.GetRoomId(), call)
}
func forwardGuard[T any](s *Server, ctx context.Context, appCode string, roomID string, call func(context.Context, roomv1.RoomGuardServiceClient) (T, error)) (T, bool, error) {
var zero T
if s == nil || s.ownerForwarder == nil {
return zero, false, nil
}
return forwardGuardToOwner(s.ownerForwarder, ctx, appCode, roomID, call)
}
func forwardQuery[T any](s *Server, ctx context.Context, appCode string, roomID string, call func(context.Context, roomv1.RoomQueryServiceClient) (T, error)) (T, bool, error) {
var zero T
if s == nil || s.ownerForwarder == nil {
return zero, false, nil
}
return forwardQueryToOwner(s.ownerForwarder, ctx, appCode, roomID, call)
}

View File

@ -2,18 +2,23 @@ package grpc
import (
"context"
"net"
"os"
"testing"
"time"
roomv1 "hyapp.local/api/proto/room/v1"
walletv1 "hyapp.local/api/proto/wallet/v1"
"hyapp/pkg/appcode"
"hyapp/pkg/xerr"
"hyapp/services/room-service/internal/integration"
roomservice "hyapp/services/room-service/internal/room/service"
"hyapp/services/room-service/internal/router"
"hyapp/services/room-service/internal/testutil/mysqltest"
grpcpkg "google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials/insecure"
grpcstatus "google.golang.org/grpc/status"
)
@ -61,3 +66,172 @@ func TestCreateRoomMapsDomainErrorToGRPCReason(t *testing.T) {
t.Fatalf("message mismatch: got %q", message)
}
}
func TestOwnerForwardingUsesRedisRouteAndNodeRegistry(t *testing.T) {
redisAddr := os.Getenv("ROOM_SERVICE_REDIS_TEST_ADDR")
if redisAddr == "" {
t.Skip("set ROOM_SERVICE_REDIS_TEST_ADDR to run Redis-backed owner forwarding test")
}
ctx := context.Background()
redisClient, err := router.NewRedisClient(ctx, redisAddr, os.Getenv("ROOM_SERVICE_REDIS_TEST_PASSWORD"), 0)
if err != nil {
t.Fatalf("open redis directory failed: %v", err)
}
t.Cleanup(func() { _ = redisClient.Close() })
directory := router.NewRedisDirectory(redisClient)
repository := mysqltest.NewRepository(t)
addrA, stopA := startForwardingRoomServer(t, "node-forward-a", directory, repository)
defer stopA()
addrB, stopB := startForwardingRoomServer(t, "node-forward-b", directory, repository)
defer stopB()
if err := directory.RegisterNode(ctx, router.NodeRegistration{NodeID: "node-forward-a", GRPCAddr: addrA}, time.Minute); err != nil {
t.Fatalf("register node a failed: %v", err)
}
if err := directory.RegisterNode(ctx, router.NodeRegistration{NodeID: "node-forward-b", GRPCAddr: addrB}, time.Minute); err != nil {
t.Fatalf("register node b failed: %v", err)
}
t.Cleanup(func() {
_ = directory.UnregisterNode(context.Background(), "node-forward-a")
_ = directory.UnregisterNode(context.Background(), "node-forward-b")
})
connA, err := grpcpkg.Dial(addrA, grpcpkg.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
t.Fatalf("dial node a failed: %v", err)
}
defer connA.Close()
connB, err := grpcpkg.Dial(addrB, grpcpkg.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
t.Fatalf("dial node b failed: %v", err)
}
defer connB.Close()
roomID := "room-forward-test"
directory.ForceExpire(roomservice.RuntimeRoomKey(appcode.Default, roomID))
t.Cleanup(func() {
directory.ForceExpire(roomservice.RuntimeRoomKey(appcode.Default, roomID))
})
createResp, err := roomv1.NewRoomCommandServiceClient(connA).CreateRoom(ctx, &roomv1.CreateRoomRequest{
Meta: &roomv1.RequestMeta{
RequestId: "req-forward-create",
CommandId: "cmd-forward-create",
ActorUserId: 1001,
RoomId: roomID,
SentAtMs: time.Now().UTC().UnixMilli(),
AppCode: appcode.Default,
},
SeatCount: 10,
Mode: "voice",
RoomName: "Forward Room",
RoomShortId: "forward-short",
})
if err != nil {
t.Fatalf("create room on node a failed: %v", err)
}
if createResp.GetRoom().GetRoomId() != roomID {
t.Fatalf("create room id mismatch: %+v", createResp.GetRoom())
}
lease, exists, err := directory.Lookup(ctx, roomservice.RuntimeRoomKey(appcode.Default, roomID))
if err != nil || !exists {
t.Fatalf("lookup owner lease failed: exists=%v err=%v", exists, err)
}
if lease.NodeID != "node-forward-a" {
t.Fatalf("room should be owned by node a before forwarding, got %+v", lease)
}
joinResp, err := roomv1.NewRoomCommandServiceClient(connB).JoinRoom(ctx, &roomv1.JoinRoomRequest{
Meta: &roomv1.RequestMeta{
RequestId: "req-forward-join",
CommandId: "cmd-forward-join",
ActorUserId: 1002,
RoomId: roomID,
SentAtMs: time.Now().UTC().UnixMilli(),
AppCode: appcode.Default,
},
Role: "audience",
})
if err != nil {
t.Fatalf("join room through node b should forward to owner: %v", err)
}
if findUser(joinResp.GetRoom(), 1002) == nil {
t.Fatalf("forwarded join did not update owner room snapshot: %+v", joinResp.GetRoom().GetOnlineUsers())
}
guardResp, err := roomv1.NewRoomGuardServiceClient(connB).CheckSpeakPermission(ctx, &roomv1.CheckSpeakPermissionRequest{
RoomId: roomID,
UserId: 1002,
AppCode: appcode.Default,
})
if err != nil {
t.Fatalf("guard through node b should forward to owner: %v", err)
}
if !guardResp.GetAllowed() {
t.Fatalf("forwarded guard should allow joined user, got %+v", guardResp)
}
snapshotResp, err := roomv1.NewRoomQueryServiceClient(connB).GetRoomSnapshot(ctx, &roomv1.GetRoomSnapshotRequest{
Meta: &roomv1.RequestMeta{
RequestId: "req-forward-snapshot",
ActorUserId: 1002,
RoomId: roomID,
AppCode: appcode.Default,
},
RoomId: roomID,
ViewerUserId: 1002,
})
if err != nil {
t.Fatalf("snapshot through node b should forward to owner: %v", err)
}
if snapshotResp.GetRoom().GetVersion() != joinResp.GetRoom().GetVersion() {
t.Fatalf("forwarded snapshot version mismatch: got %d want %d", snapshotResp.GetRoom().GetVersion(), joinResp.GetRoom().GetVersion())
}
}
func startForwardingRoomServer(t *testing.T, nodeID string, directory *router.RedisDirectory, repository roomservice.Repository) (string, func()) {
t.Helper()
svc := roomservice.New(roomservice.Config{
NodeID: nodeID,
LeaseTTL: 10 * time.Second,
RankLimit: 20,
SnapshotEveryN: 1,
}, directory, repository, fakeWallet{}, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher())
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen forwarding test server failed: %v", err)
}
grpcServer := grpcpkg.NewServer()
roomServer := NewServer(svc, WithOwnerForwarding(nodeID, directory, directory, time.Second))
roomv1.RegisterRoomCommandServiceServer(grpcServer, roomServer)
roomv1.RegisterRoomGuardServiceServer(grpcServer, roomServer)
roomv1.RegisterRoomQueryServiceServer(grpcServer, roomServer)
errCh := make(chan error, 1)
go func() {
errCh <- grpcServer.Serve(listener)
}()
stop := func() {
roomServer.ownerForwarder.close()
grpcServer.GracefulStop()
select {
case err := <-errCh:
if err != nil {
t.Fatalf("forwarding test server stopped with error: %v", err)
}
case <-time.After(2 * time.Second):
t.Fatal("forwarding test server did not stop")
}
}
return listener.Addr().String(), stop
}
func findUser(snapshot *roomv1.RoomSnapshot, userID int64) *roomv1.RoomUser {
for _, user := range snapshot.GetOnlineUsers() {
if user.GetUserId() == userID {
return user
}
}
return nil
}

View File

@ -30,7 +30,7 @@ COPY services/user-service/configs/config.docker.yaml /app/config.yaml
USER appuser
EXPOSE 13005
EXPOSE 13005 13105
ENTRYPOINT ["/app/server"]
CMD ["-config", "/app/config.yaml"]

View File

@ -8,6 +8,7 @@ log:
include_response_body: false
max_payload_bytes: 2048
grpc_addr: ":13005"
health_http_addr: ":13105"
mysql_dsn: "hyapp:hyapp@tcp(mysql:3306)/hyapp_user?parseTime=true&charset=utf8mb4&loc=UTC"
mysql_auto_migrate: false
id_generator:

View File

@ -8,6 +8,7 @@ log:
include_response_body: false
max_payload_bytes: 2048
grpc_addr: ":13005"
health_http_addr: ":13105"
mysql_dsn: "USER:PASSWORD@tcp(TENCENT_CDB_HOST:3306)/hyapp_user?parseTime=true&charset=utf8mb4&loc=UTC&tls=false"
mysql_auto_migrate: false
id_generator:

View File

@ -8,6 +8,7 @@ log:
include_response_body: false
max_payload_bytes: 2048
grpc_addr: ":13005"
health_http_addr: ":13105"
mysql_dsn: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_user?parseTime=true&charset=utf8mb4&loc=UTC"
mysql_auto_migrate: false
id_generator:

View File

@ -11,6 +11,8 @@ import (
userv1 "hyapp.local/api/proto/user/v1"
"hyapp/pkg/grpchealth"
"hyapp/pkg/grpcshutdown"
"hyapp/pkg/healthhttp"
"hyapp/pkg/idgen"
"hyapp/pkg/logx"
"hyapp/services/user-service/internal/config"
@ -36,6 +38,8 @@ type App struct {
listener net.Listener
// health 是标准 gRPC health 的 serving/draining 状态来源。
health *grpchealth.ServingChecker
// healthHTTP 给 CLB 和发布脚本提供 HTTP readiness不承载业务请求。
healthHTTP *healthhttp.Server
// mysqlRepo 持有用户主数据、认证身份、session 和短号事务的唯一运行时存储。
mysqlRepo *mysqlstorage.Repository
// walletDB 只用于读取 wallet_outbox 充值事实,实际账务状态仍由 wallet-service owner。
@ -226,15 +230,47 @@ func New(cfg config.Config) (*App, error) {
userv1.RegisterUserHostAdminServiceServer(server, userServer)
userv1.RegisterUserCronServiceServer(server, userServer)
health := grpchealth.NewServingChecker("user-service")
healthDependencies := []grpchealth.Dependency{{
Name: "mysql",
Check: mysqlRepo.Ping,
}}
if walletDB != nil {
healthDependencies = append(healthDependencies, grpchealth.Dependency{
Name: "wallet_mysql",
Check: walletDB.PingContext,
})
}
if roomDB != nil {
healthDependencies = append(healthDependencies, grpchealth.Dependency{
Name: "room_mysql",
Check: roomDB.PingContext,
})
}
health := grpchealth.NewServingChecker("user-service", healthDependencies...)
// user-service 目前使用简单 serving checker存储探测由后续专用 health 可扩展。
healthgrpc.RegisterHealthServer(server, grpchealth.NewServer(health))
healthHTTP, err := healthhttp.New(cfg.HealthHTTPAddr, cfg.NodeID, health)
if err != nil {
if roomDB != nil {
_ = roomDB.Close()
}
if walletDB != nil {
_ = walletDB.Close()
}
if loginRiskRedisClose != nil {
_ = loginRiskRedisClose()
}
_ = listener.Close()
_ = mysqlRepo.Close()
return nil, err
}
workerCtx, workerCancel := context.WithCancel(context.Background())
return &App{
server: server,
listener: listener,
health: health,
healthHTTP: healthHTTP,
mysqlRepo: mysqlRepo,
walletDB: walletDB,
roomDB: roomDB,
@ -251,6 +287,7 @@ func New(cfg config.Config) (*App, error) {
// Run 启动 gRPC 服务。
func (a *App) Run() error {
a.runHealthHTTP()
// 只有 listener 进入 Serve 后才标记 serving避免健康检查提前放行。
a.health.MarkServing()
defer func() {
@ -297,7 +334,8 @@ func (a *App) Close() {
}
a.workerWG.Wait()
// GracefulStop 等待已进入的 RPC 结束,避免 token/session 写入被硬切。
a.server.GracefulStop()
grpcshutdown.GracefulStop(a.server, 15*time.Second)
a.closeHealthHTTP()
if a.mysqlRepo != nil {
// MySQL 连接池最后关闭,保证 GracefulStop 期间 repository 仍可用。
_ = a.mysqlRepo.Close()
@ -313,3 +351,23 @@ func (a *App) Close() {
}
})
}
func (a *App) runHealthHTTP() {
if a.healthHTTP == nil {
return
}
go func() {
if err := a.healthHTTP.Run(); err != nil {
logx.Error(context.Background(), "health_http_run_failed", err)
}
}()
}
func (a *App) closeHealthHTTP() {
if a.healthHTTP == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
_ = a.healthHTTP.Close(ctx)
}

View File

@ -18,6 +18,8 @@ type Config struct {
Environment string `yaml:"environment"`
// GRPCAddr 是 user-service 内部 gRPC 监听地址。
GRPCAddr string `yaml:"grpc_addr"`
// HealthHTTPAddr 是给 CLB/发布脚本探活使用的 HTTP 监听地址,不承载业务流量。
HealthHTTPAddr string `yaml:"health_http_addr"`
// MySQLDSN 是用户主数据、认证身份、session 和短号事务的必需存储连接串。
MySQLDSN string `yaml:"mysql_dsn"`
// MySQLAutoMigrate 保留给显式本地迁移;线上 schema 由发布流程或 initdb 管理。
@ -132,6 +134,7 @@ func Default() Config {
NodeID: "user-local",
Environment: "local",
GRPCAddr: ":13005",
HealthHTTPAddr: ":13105",
MySQLDSN: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_user?parseTime=true&charset=utf8mb4&loc=UTC",
MySQLAutoMigrate: false,
IDGenerator: IDGeneratorConfig{
@ -209,6 +212,10 @@ func Load(path string) (Config, error) {
if cfg.Environment == "" {
cfg.Environment = "local"
}
cfg.HealthHTTPAddr = strings.TrimSpace(cfg.HealthHTTPAddr)
if cfg.HealthHTTPAddr == "" {
cfg.HealthHTTPAddr = ":13105"
}
cfg.LoginRisk.RedisAddr = strings.TrimSpace(cfg.LoginRisk.RedisAddr)
if cfg.LoginRisk.BlockedTTLSec <= 0 {
cfg.LoginRisk.BlockedTTLSec = 2592000

File diff suppressed because it is too large Load Diff

View File

@ -6,11 +6,12 @@ import (
"errors"
"strings"
mysqldriver "github.com/go-sql-driver/mysql"
"hyapp/pkg/appcode"
"hyapp/pkg/xerr"
userdomain "hyapp/services/user-service/internal/domain/user"
"hyapp/services/user-service/internal/storage/mysql/shared"
mysqldriver "github.com/go-sql-driver/mysql"
)
func (r *Repository) ResolveEnabledCountryByCode(ctx context.Context, countryCode string) (userdomain.Country, bool, error) {
@ -20,7 +21,8 @@ func (r *Repository) ResolveEnabledCountryByCode(ctx context.Context, countryCod
// ListRegistrationCountries 返回注册页可选国家,只包含 enabled 的国家。
func (r *Repository) ListRegistrationCountries(ctx context.Context) ([]userdomain.Country, error) {
return listCountries(ctx, r.db, userdomain.CountryFilter{Enabled: new(true)})
enabled := true
return listCountries(ctx, r.db, userdomain.CountryFilter{Enabled: &enabled})
}
// ResolveActiveRegionByCountry 按国家码读取当前 active 区域映射;无显式映射时由 regions.go 返回 GLOBAL。
@ -130,11 +132,6 @@ func scanCountry(scanner interface{ Scan(dest ...any) error }) (userdomain.Count
return country, err
}
//go:fix inline
func boolPtr(value bool) *bool {
return new(value)
}
func canonicalCountryCodes(ctx context.Context, q queryer, countries []string) ([]string, error) {
seen := make(map[string]struct{}, len(countries))
result := make([]string, 0, len(countries))

View File

@ -30,7 +30,7 @@ COPY services/wallet-service/configs/config.docker.yaml /app/config.yaml
USER appuser
EXPOSE 13004
EXPOSE 13004 13104
ENTRYPOINT ["/app/server"]
CMD ["-config", "/app/config.yaml"]

View File

@ -8,5 +8,6 @@ log:
include_response_body: false
max_payload_bytes: 2048
grpc_addr: ":13004"
health_http_addr: ":13104"
mysql_dsn: "hyapp:hyapp@tcp(mysql:3306)/hyapp_wallet?parseTime=true&charset=utf8mb4&loc=UTC"
mysql_auto_migrate: false

View File

@ -8,5 +8,6 @@ log:
include_response_body: false
max_payload_bytes: 2048
grpc_addr: ":13004"
health_http_addr: ":13104"
mysql_dsn: "USER:PASSWORD@tcp(TENCENT_CDB_HOST:3306)/hyapp_wallet?parseTime=true&charset=utf8mb4&loc=UTC&tls=false"
mysql_auto_migrate: false

View File

@ -8,5 +8,6 @@ log:
include_response_body: false
max_payload_bytes: 2048
grpc_addr: ":13004"
health_http_addr: ":13104"
mysql_dsn: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_wallet?parseTime=true&charset=utf8mb4&loc=UTC"
mysql_auto_migrate: false

View File

@ -11,6 +11,8 @@ import (
healthgrpc "google.golang.org/grpc/health/grpc_health_v1"
walletv1 "hyapp.local/api/proto/wallet/v1"
"hyapp/pkg/grpchealth"
"hyapp/pkg/grpcshutdown"
"hyapp/pkg/healthhttp"
"hyapp/pkg/logx"
"hyapp/services/wallet-service/internal/config"
walletservice "hyapp/services/wallet-service/internal/service/wallet"
@ -20,11 +22,12 @@ import (
// App 装配 wallet-service gRPC 入口。
type App struct {
server *grpc.Server
listener net.Listener
health *grpchealth.ServingChecker
mysqlRepo *mysqlstorage.Repository
closeOnce sync.Once
server *grpc.Server
listener net.Listener
health *grpchealth.ServingChecker
healthHTTP *healthhttp.Server
mysqlRepo *mysqlstorage.Repository
closeOnce sync.Once
}
// New 初始化 wallet-service。
@ -47,19 +50,30 @@ func New(cfg config.Config) (*App, error) {
server := grpc.NewServer(grpc.UnaryInterceptor(logx.UnaryServerInterceptor("wallet-service")))
svc := walletservice.New(repository)
walletv1.RegisterWalletServiceServer(server, grpcserver.NewServer(svc))
health := grpchealth.NewServingChecker("wallet-service")
health := grpchealth.NewServingChecker("wallet-service", grpchealth.Dependency{
Name: "mysql",
Check: repository.Ping,
})
healthgrpc.RegisterHealthServer(server, grpchealth.NewServer(health))
healthHTTP, err := healthhttp.New(cfg.HealthHTTPAddr, cfg.NodeID, health)
if err != nil {
_ = listener.Close()
_ = repository.Close()
return nil, err
}
return &App{
server: server,
listener: listener,
health: health,
mysqlRepo: repository,
server: server,
listener: listener,
health: health,
healthHTTP: healthHTTP,
mysqlRepo: repository,
}, nil
}
// Run 启动 gRPC 服务。
func (a *App) Run() error {
a.runHealthHTTP()
a.health.MarkServing()
err := a.server.Serve(a.listener)
a.health.MarkStopped()
@ -74,10 +88,31 @@ func (a *App) Run() error {
func (a *App) Close() {
a.closeOnce.Do(func() {
a.health.MarkDraining()
a.server.GracefulStop()
grpcshutdown.GracefulStop(a.server, 15*time.Second)
a.closeHealthHTTP()
if a.mysqlRepo != nil {
// MySQL 连接池最后关闭,避免正在 drain 的扣费请求丢失提交能力。
_ = a.mysqlRepo.Close()
}
})
}
func (a *App) runHealthHTTP() {
if a.healthHTTP == nil {
return
}
go func() {
if err := a.healthHTTP.Run(); err != nil {
logx.Error(context.Background(), "health_http_run_failed", err)
}
}()
}
func (a *App) closeHealthHTTP() {
if a.healthHTTP == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
_ = a.healthHTTP.Close(ctx)
}

View File

@ -13,6 +13,8 @@ type Config struct {
NodeID string `yaml:"node_id"`
Environment string `yaml:"environment"`
GRPCAddr string `yaml:"grpc_addr"`
// HealthHTTPAddr 是给 CLB/发布脚本探活使用的 HTTP 监听地址,不承载业务流量。
HealthHTTPAddr string `yaml:"health_http_addr"`
// MySQLDSN 是账户、交易、分录和 outbox 的必需存储连接串。
MySQLDSN string `yaml:"mysql_dsn"`
// MySQLAutoMigrate 保留给显式本地迁移;线上 schema 由发布流程或 initdb 管理。
@ -27,6 +29,7 @@ func Default() Config {
NodeID: "wallet-local",
Environment: "local",
GRPCAddr: ":13004",
HealthHTTPAddr: ":13104",
MySQLDSN: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_wallet?parseTime=true&charset=utf8mb4&loc=UTC",
MySQLAutoMigrate: false,
Log: logx.Config{
@ -60,6 +63,10 @@ func Load(path string) (Config, error) {
if cfg.Environment == "" {
cfg.Environment = "local"
}
cfg.HealthHTTPAddr = strings.TrimSpace(cfg.HealthHTTPAddr)
if cfg.HealthHTTPAddr == "" {
cfg.HealthHTTPAddr = ":13104"
}
return cfg, nil
}