From 85ff03764702e4ba15b833d45562074b9f338577 Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Wed, 15 Jul 2026 18:30:02 +0800 Subject: [PATCH] fix: restore active channels after gateway restart --- api/src/channels/channels.service.spec.ts | 20 +++++++++ api/src/channels/channels.service.ts | 41 +++++++++++++++---- .../first-version-development-requirements.md | 1 + docs/production-deployment.md | 3 ++ docs/system-functional-test-cases.md | 1 + docs/testing-progress.md | 4 +- tools/deploy/production-deploy.sh | 2 +- 7 files changed, 61 insertions(+), 11 deletions(-) diff --git a/api/src/channels/channels.service.spec.ts b/api/src/channels/channels.service.spec.ts index 722e55b..4d1c585 100644 --- a/api/src/channels/channels.service.spec.ts +++ b/api/src/channels/channels.service.spec.ts @@ -286,6 +286,26 @@ describe('ChannelsService', () => { expect(prisma.smsChannel.create).not.toHaveBeenCalled(); }); + it('replays every active channel connection after Gateway restart', async () => { + const prisma = createPrismaMock(); + const activeChannel = await prisma.smsChannel.findUnique(); + prisma.smsChannel.findMany.mockResolvedValue([activeChannel]); + const service = new ChannelsService(prisma as never); + + await (service as unknown as { reconnectActiveChannelsAfterGatewayRestart(): Promise }) + .reconnectActiveChannelsAfterGatewayRestart(); + + expect(prisma.smsChannel.findMany).toHaveBeenCalledWith({ where: { status: 'active' } }); + expect(mockQueueAdd).toHaveBeenCalledWith('connect-channel', expect.objectContaining({ + channelId: 'channel-1', + reason: 'gateway_restarted', + channel: expect.objectContaining({ rateLimitPerSecond: 100 }), + }), { jobId: 'channel-1:primary:connect' }); + expect(mockFetch).toHaveBeenCalledWith('http://127.0.0.1:8090/connections/connect', expect.objectContaining({ + body: expect.stringContaining('"reason":"gateway_restarted"'), + })); + }); + it('creates CMPP channels and route rules with first-version defaults', async () => { const prisma = createPrismaMock(); const service = new ChannelsService(prisma as never); diff --git a/api/src/channels/channels.service.ts b/api/src/channels/channels.service.ts index 8ccf244..3b22eb2 100644 --- a/api/src/channels/channels.service.ts +++ b/api/src/channels/channels.service.ts @@ -180,6 +180,7 @@ const DEFAULT_GATEWAY_CONTROL_URL = 'http://127.0.0.1:8090'; const DEFAULT_CHANNEL_CONNECTION_ID = 'primary'; const DEFAULT_CONNECTING_TIMEOUT_MS = 30_000; const DEFAULT_CONNECTING_TIMEOUT_SCAN_MS = 5_000; +const DEFAULT_GATEWAY_STARTUP_RECONNECT_DELAY_MS = 1_000; const CONNECTING_TIMEOUT_ERROR = 'Gateway connection request timed out'; const DEFAULT_CMPP_VERSION = '2.0'; @@ -190,25 +191,32 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy { private gatewaySubmitQueue?: Queue; private redis?: IORedis; private connectionTimeoutTimer?: ReturnType; + private gatewayStartupReconnectTimer?: ReturnType; constructor(private readonly prisma: PrismaService) {} onModuleInit() { - if (process.env.GATEWAY_CONNECTING_TIMEOUT_SCANNER_DISABLED === 'true') { - return; + if (process.env.GATEWAY_CONNECTING_TIMEOUT_SCANNER_DISABLED !== 'true') { + this.connectionTimeoutTimer = setInterval(() => { + void this.markTimedOutConnectingChannels().catch((error) => { + this.logger.error(`Failed to mark timed-out CMPP connections: ${error instanceof Error ? error.message : String(error)}`); + }); + }, getPositiveIntegerEnv('GATEWAY_CONNECTING_TIMEOUT_SCAN_MS', DEFAULT_CONNECTING_TIMEOUT_SCAN_MS)); + this.connectionTimeoutTimer.unref?.(); } - this.connectionTimeoutTimer = setInterval(() => { - void this.markTimedOutConnectingChannels().catch((error) => { - this.logger.error(`Failed to mark timed-out CMPP connections: ${error instanceof Error ? error.message : String(error)}`); - }); - }, getPositiveIntegerEnv('GATEWAY_CONNECTING_TIMEOUT_SCAN_MS', DEFAULT_CONNECTING_TIMEOUT_SCAN_MS)); - this.connectionTimeoutTimer.unref?.(); + this.gatewayStartupReconnectTimer = setTimeout(() => { + void this.reconnectActiveChannelsAfterGatewayRestart(); + }, getPositiveIntegerEnv('GATEWAY_STARTUP_RECONNECT_DELAY_MS', DEFAULT_GATEWAY_STARTUP_RECONNECT_DELAY_MS)); + this.gatewayStartupReconnectTimer.unref?.(); } async onModuleDestroy() { if (this.connectionTimeoutTimer) { clearInterval(this.connectionTimeoutTimer); } + if (this.gatewayStartupReconnectTimer) { + clearTimeout(this.gatewayStartupReconnectTimer); + } await this.gatewayConnectionQueue?.close(); await this.gatewaySubmitQueue?.close(); this.redis?.disconnect(); @@ -1232,7 +1240,7 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy { rateLimitPerSecond: number; config?: Prisma.JsonValue | null; }, - reason: 'channel_created' | 'channel_enabled', + reason: 'channel_created' | 'channel_enabled' | 'gateway_restarted', operatorId?: string, ) { const desiredConnections = getDesiredConnections(channel.config); @@ -1300,6 +1308,21 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy { return state; } + private async reconnectActiveChannelsAfterGatewayRestart() { + const channels = await this.prisma.smsChannel.findMany({ where: { status: 'active' } }); + const results = await Promise.allSettled( + channels.map((channel) => this.requestChannelConnection(channel, 'gateway_restarted')), + ); + results.forEach((result, index) => { + if (result.status === 'rejected') { + const channel = channels[index]; + this.logger.error( + `Failed to restore active CMPP channel ${channel?.code ?? channel?.id ?? index}: ${result.reason instanceof Error ? result.reason.message : String(result.reason)}`, + ); + } + }); + } + private getGatewayConnectionQueue() { this.gatewayConnectionQueue ??= new Queue(GATEWAY_CONNECTION_QUEUE, { connection: bullmqConnection() }); return this.gatewayConnectionQueue; diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index e0b19b7..ef612d5 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -281,6 +281,7 @@ - 已实现“上游可能已受理但 submit resp 丢失”场景的保守补偿第一版：Gateway 在 receipt 事件中补充手机号；NestJS 对无法按 `messageId/gatewayMessageId` 精确命中的回执，只在“同通道、同手机号、72 小时窗口内、且仅存在 1 条 `timeout + gatewayMessageId=null` 的 submit 记录”时才回填并接收该回执，避免误绑到其他短信。 - 已实现 Gateway 提交异常治理：Go Gateway 对多次处理仍失败的 `SubmitCommand` 不再无限滞留在 PEL，而是按阈值写入 NestJS 真实 `GatewaySubmitDeadLetter` 表（数据库表名和内部接口保留技术兼容名，页面统一称“Gateway提交异常”）。运营端 `/admin/gateway-submit-exceptions` 提供真实分页、筛选、汇总、脱敏详情和单条重新入队；原始 payload、密码、密钥不得返回浏览器。重新入队必须要求近期认证、填写原因、勾选“已确认上游未受理”，并校验短信尚未 accepted/submitted/delivered/unknown、通道 active 且 connected、人工次数小于 3；服务端以 pending 到 requeueing 的原子状态抢占防止重复点击，成功写回 Redis Stream 后记录操作人、原因、Stream ID 和时间。收到后续 SubmitResult 时必须将对应异常记录闭环为 resolved。 - 已实现 Gateway 通道级 Redis 限速：NestJS 入队前保留业务层通道限速，Go Gateway 在真正调用上游 Submit 前再次按通道 ID 预约发送时隙；连接命令把权威 TPS 写入 Redis，提交按权威值与消息值的较小者执行。普通 Stream 消息在等待期间不 ACK、不转失败，多实例共同使用同一限速状态；worker 对同批消息并发调度，低 TPS 通道等待不阻塞其他通道。 +- 已实现 Gateway 重启后的 active 上游通道恢复：部署先重启 Gateway 再重启 API；API 启动后从 PostgreSQL 读取 active 通道，重新下发连接命令，恢复 Gateway 内存连接池、真实连接状态和 Redis 权威 TPS key，不得继续沿用重启前的 connected 状态。 - 已实现客户侧下游投递重试第二版：客户系统负责断线后重连；平台在客户离线或投递失败时把 Deliver Receipt/上行 Deliver 保留在 `CmppDownstreamDelivery`，客户 bind 成功后立即拉取 pending，且 Gateway 会对当前在线账号周期补投；超过重试上限后转 `failed` 并写失败审计。 - 已实现下游投递失败审计与人工重投第一版：运营端后端与页面可分页查看 `CmppDownstreamDelivery` 的 pending/awaiting_ack/failed/unconfirmed/rejected/delivered 记录，支持按状态、类型、应用和关键字筛选，并可对非 `awaiting_ack` 记录执行人工重投，真实调用 Gateway `/downstream/receipt` 或 `/downstream/uplink`。主记录必须分开保存自动重试次数 `retryCount`、人工重投次数 `manualRetryCount` 和最近人工重投时间 `lastRetriedAt`，操作日志保留重投前状态与自动重试次数。 - 已实现下游投递批量重投第一版：运营端可在当前页勾选多条 `pending/failed` 下游投递记录，调用真实批量接口逐条重投并返回成功/失败汇总，不允许用前端循环假装成功。 diff --git a/docs/production-deployment.md b/docs/production-deployment.md index f07c165..cdbcf30 100644 --- a/docs/production-deployment.md +++ b/docs/production-deployment.md @@ -49,6 +49,7 @@ REPORT_DAILY_REFRESH_ENABLED=true REPORT_REFRESH_INTERVAL_MS=3600000 CMPP_DOWNSTREAM_ACK_TIMEOUT_SECONDS=30 GATEWAY_CMPP_ADDR=0.0.0.0:17890 +GATEWAY_STARTUP_RECONNECT_DELAY_MS=1000 OBJECT_STORAGE_DRIVER=minio OBJECT_STORAGE_LOCAL_ROOT=/var/lib/cmpp-platform/object-storage PROD_ADMIN_EMAIL=admin@example.com @@ -66,6 +67,8 @@ PROD_ADMIN_PASSWORD='change-me' Gateway 的最终 TPS 防线依赖与 API 相同的 Redis。通道连接时会写入 `rate:gateway:channel:config:` 权威上限，实际预约使用 `rate:gateway:channel:`；这些 key 不应在正常发布时清理。多 Gateway 实例必须指向同一 Redis，才能共享单通道额度。超速的 `gateway.submit.commands` 消息会保持在 consumer group pending 中等待，不应通过手工 `XACK` 或删除 Stream 处理积压；先检查通道配置、Redis key、consumer group 和 Gateway 日志。 +服务重启顺序必须是 Gateway 在前、API 在后。API 启动后等待 `GATEWAY_STARTUP_RECONNECT_DELAY_MS`（默认 1 秒），从 PostgreSQL 读取全部 active 通道并重新下发真实连接命令，同时恢复 Gateway 内存连接池和 Redis 权威 TPS key；禁止沿用数据库中重启前的 connected 状态冒充当前连接。 + 日报任务默认启用，并由 `REPORT_REFRESH_INTERVAL_MS` 每小时检查一次北京时间业务日是否变化；每个业务日只执行一次 T-4 至 T-1 重算。服务重启后也会自动补跑最近四个完整自然日，确保 72 小时回执更新反映到对账和利润报表。 如生产验证服务器临时无法稳定下载 MinIO，可显式传入 `OBJECT_STORAGE_DRIVER=local`，文件会通过真实 API 保存到服务器本地目录 `OBJECT_STORAGE_LOCAL_ROOT`，`cmpp-minio` 服务会跳过安装和启动。该模式只建议用于验证环境；正式生产建议恢复 `OBJECT_STORAGE_DRIVER=minio`。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index 7eb066f..23744ed 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -3452,3 +3452,4 @@ npm run verify:phase8 | TC-GW-RATE-001 | 给通道 A 配置 10 TPS，连续投递 20 条；通道 B 同时配置 20 TPS 并投递，另让提交命令携带高于通道配置的数值。 | Gateway A 实际提交节奏不超过 10 TPS，B 独立按自身额度执行；消息值不能放大 A 的权威上限，同一通道跨通道组共享额度。 | | TC-GW-RATE-002 | 超过通道 TPS 后观察 Redis Stream consumer group，并在存在等待消息时重启 Gateway。 | 超流速消息保留在 Stream pending，不直接失败；重启后通过 PEL/XAUTOCLAIM 恢复并继续按通道 TPS 排队提交，不丢失、不重复 ACK。 | | TC-GW-RATE-003 | 启动两个共享同一 Redis 的 Gateway 消费实例，同时向同一通道发送，再向两个不同通道发送。 | 同一通道的两个实例共享 Redis 限速额度，总 TPS 不叠加；不同通道使用独立 key，不被合并成平台总 TPS。 | +| TC-GW-RATE-004 | 在存在 active 上游通道时按生产脚本顺序重启 Gateway 和 API，随后检查 Gateway 日志、连接状态与 Redis。 | Gateway 先启动，API 随后重新下发全部 active 通道连接命令；Gateway 内存连接池恢复，数据库状态反映本次真实连接结果，并生成 `rate:gateway:channel:config:`，不沿用重启前的假 connected。 | diff --git a/docs/testing-progress.md b/docs/testing-progress.md index 1bb4943..1b07307 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -1936,4 +1936,6 @@ git diff --check - 运营端新增“Gateway提交异常”页面和真实 NestJS API：支持状态/应用/通道/关键字筛选、服务端汇总、脱敏详情及单条重新入队；浏览器响应不包含原始 payload、通道密码、密钥或鉴权字段，数据库和内部兼容接口暂保留 `GatewaySubmitDeadLetter` 技术命名。 - 重新入队增加近期认证、人工原因、明确确认上游未受理、短信状态、通道状态/真实连接状态、最多 3 次以及 pending→requeueing 原子抢占校验；写入 Redis Stream 失败会恢复 pending，成功记录操作人、原因、Stream ID 和时间，后续 SubmitResult 自动闭环 resolved。 - Go Gateway 新增 Redis 分布式单通道限速。连接命令保存权威通道 TPS，提交取权威值与消息值的较小者；同一通道在多实例和多个通道组间共享额度，不同通道独立。Stream 消息超速时保持未 ACK 并等待，不作为发送失败，重启后继续使用既有 pending 恢复机制；worker 对同批消息并发处理，避免低 TPS 通道等待阻塞其他通道。 -- 定向验证已通过：Gateway `internal/ratelimit`、`internal/control`、`internal/submitworker`；API `operations.service.spec.ts`、`send-chain.service.spec.ts` 共 73 项。最终本地门禁通过：Prisma validate/generate、51 条 migration status、API 19 suites/199 项、API build、前端 build、Gateway `go test ./...` 和 `git diff --check`；前端仅有既有 chunk size warning，Jest 仍需 `--forceExit` 退出既有异步句柄。应用内浏览器确认受保护新路由真实跳转运营端登录、标题和 DOM 正常、console 无 error/warn；因真实图形验证码未获授权代解，未绕过认证或注入会话。生产备份、提交、push 与部署结果在发布后补录。 +- 定向验证已通过：Gateway `internal/ratelimit`、`internal/control`、`internal/submitworker`；API `operations.service.spec.ts`、`send-chain.service.spec.ts` 共 73 项。最终本地门禁通过：Prisma validate/generate、51 条 migration status、API 19 suites/200 项、API build、前端 build、Gateway `go test ./...` 和 `git diff --check`；前端仅有既有 chunk size warning，Jest 仍需 `--forceExit` 退出既有异步句柄。应用内浏览器确认受保护新路由真实跳转运营端登录、标题和 DOM 正常、console 无 error/warn；因真实图形验证码未获授权代解，未绕过认证或注入会话。 +- 首轮生产部署后发现 active 通道在数据库仍显示旧 connected，但新 Gateway 进程没有内存连接池且 Redis 权威 TPS key 为空。根因为发布脚本先重启 API、后重启 Gateway，并且 API 启动未重放 active 通道。已修复为先 Gateway 后 API，API 启动延迟 1 秒后从真实数据库重新下发全部 active 通道；复验结果随最终发布补录。 +- 功能提交 `7091a8be` 已 push 并完成首轮部署；部署前备份 PostgreSQL、运行源码和环境文件至 `/opt/cmpp-platform/backups/releases/20260715-182455`，三份备份均通过完整性与 SHA-256 检查。首轮发布包本地/服务器 SHA-256 均为 `dc9fce751fce42bcf8ac14a4b9fd6a1d28a489469350904ee395db3ceb9dcab9`，4 条 migration 成功应用，51 条齐全；首轮健康检查、端口、受保护异常 API 401 和静态资源均通过。由于随后发现并修复上述 Gateway 重启恢复缺口，最终运行提交和复验结果以修正发布记录为准。 diff --git a/tools/deploy/production-deploy.sh b/tools/deploy/production-deploy.sh index 5811d31..62f6817 100644 --- a/tools/deploy/production-deploy.sh +++ b/tools/deploy/production-deploy.sh @@ -63,8 +63,8 @@ else systemctl restart cmpp-minio systemctl enable --now cmpp-api cmpp-gateway nginx fi -systemctl restart cmpp-api systemctl restart cmpp-gateway +systemctl restart cmpp-api systemctl restart nginx echo "[deploy] Health checks"