From 76e7c8c401e55b19ed8a67272b770395749a022b Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Tue, 25 Aug 2026 17:11:55 +0800 Subject: [PATCH] perf: batch inbound workflows by tenant --- api/src/send-chain/send-chain.service.spec.ts | 71 +++++++++++++++- .../send-chain/send-gateway-submit.service.ts | 22 ++--- .../send-chain/send-inbound-entry.service.ts | 84 ++++++++++++++++--- .../first-version-development-requirements.md | 8 ++ ...phase-5-gateway-capacity-expansion-plan.md | 6 ++ docs/system-functional-test-cases.md | 4 + tools/deploy/production-bootstrap.sh | 4 + tools/deploy/production-deploy.sh | 8 ++ tools/deploy/verify-production-deployment.mjs | 2 + 9 files changed, 184 insertions(+), 25 deletions(-) diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index 4cc6a2f..a26495f 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -2449,11 +2449,80 @@ describe('SendChainService', () => { const { service, prisma } = createService(); prisma.$queryRaw.mockResolvedValueOnce([]); - await (service as any).submission.inboundEntry.claimInboundWorkflows(5); + await (service as any).submission.inboundEntry.claimInboundWorkflows(5, ['tenant-busy']); const sql = prisma.$queryRaw.mock.calls[0][0].strings.join(' '); expect(sql).toContain("NOW() AT TIME ZONE 'UTC'"); + expect(sql).toContain('"tenantId" NOT IN'); + expect(prisma.$queryRaw.mock.calls[0][0].values).toContain('tenant-busy'); expect(sql).toContain('FOR UPDATE SKIP LOCKED'); + expect(sql).toContain('inbox."tenantId"'); + }); + + it('groups Inbox work by tenant and keeps chunks for one tenant serial', async () => { + const { service } = createService(); + const inboundEntry = (service as any).submission.inboundEntry; + const items = [ + { id: 'a1', tenantId: 'tenant-a', applicationId: 'app-1' }, + { id: 'b1', tenantId: 'tenant-b', applicationId: 'app-2' }, + { id: 'a2', tenantId: 'tenant-a', applicationId: 'app-1' }, + { id: 'a3', tenantId: 'tenant-a', applicationId: 'app-1' }, + ]; + const grouped = inboundEntry.groupInboundWorkflowsByTenant(items); + expect([...grouped.keys()]).toEqual(['tenant-a', 'tenant-b']); + expect(grouped.get('tenant-a').map((item: { id: string }) => item.id)).toEqual(['a1', 'a2', 'a3']); + + inboundEntry.processClaimedInboundWorkflowBatch = jest.fn().mockResolvedValue(undefined); + await inboundEntry.processTenantInboundWorkflowBatches(grouped.get('tenant-a'), 2, new Map()); + expect(inboundEntry.processClaimedInboundWorkflowBatch.mock.calls.map((call: unknown[]) => ( + (call[0] as Array<{ id: string }>).map((item) => item.id) + ))).toEqual([['a1', 'a2'], ['a3']]); + }); + + it('coalesces a small ready Inbox set before claiming the next tenant batch', async () => { + const previousWait = process.env.API_INBOUND_WORKFLOW_BATCH_WAIT_MS; + const previousTarget = process.env.API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE; + process.env.API_INBOUND_WORKFLOW_BATCH_WAIT_MS = '5'; + process.env.API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE = '32'; + try { + const { service, prisma } = createService(); + prisma.$queryRaw.mockResolvedValueOnce([{ count: 3n }]); + const startedAt = Date.now(); + await (service as any).submission.inboundEntry.waitForInboundWorkflowMicroBatch(96, ['tenant-busy']); + expect(Date.now() - startedAt).toBeGreaterThanOrEqual(4); + const query = prisma.$queryRaw.mock.calls[0][0]; + expect(query.strings.join(' ')).toContain('COUNT(*)::bigint'); + expect(query.values).toContain('tenant-busy'); + } finally { + if (previousWait == null) delete process.env.API_INBOUND_WORKFLOW_BATCH_WAIT_MS; + else process.env.API_INBOUND_WORKFLOW_BATCH_WAIT_MS = previousWait; + if (previousTarget == null) delete process.env.API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE; + else process.env.API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE = previousTarget; + } + }); + + it('uses UTC for Submit Outbox claim, lease, publish, and retry timestamps', async () => { + const previousShadow = process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED; + const previousPublish = process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED; + process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED = 'true'; + delete process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED; + try { + const { service, prisma } = createService(); + prisma.$queryRaw.mockResolvedValueOnce([{ id: 'outbox-1', submitId: 'SUB-1', payload: { messageId: 'MSG-1' } }]); + const gatewaySubmit = (service as any).submission.gatewaySubmit; + await gatewaySubmit.publishSubmitOutboxBatch(); + + const claimSql = prisma.$queryRaw.mock.calls[0][0].strings.join(' '); + const publishSql = prisma.$executeRaw.mock.calls.at(-1)[0].strings.join(' '); + expect(claimSql).toContain("NOW() AT TIME ZONE 'UTC'"); + expect(publishSql).toContain("NOW() AT TIME ZONE 'UTC'"); + expect(`${claimSql} ${publishSql}`).not.toContain('CURRENT_TIMESTAMP'); + } finally { + if (previousShadow == null) delete process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED; + else process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED = previousShadow; + if (previousPublish == null) delete process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED; + else process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED = previousPublish; + } }); it('enqueues a freshly persisted inbound message without querying the task and message again', async () => { diff --git a/api/src/send-chain/send-gateway-submit.service.ts b/api/src/send-chain/send-gateway-submit.service.ts index 6c0b201..e02e32c 100644 --- a/api/src/send-chain/send-gateway-submit.service.ts +++ b/api/src/send-chain/send-gateway-submit.service.ts @@ -298,7 +298,7 @@ startSubmitOutboxPublisher() { "receiptStatus" = NULL, "errorCode" = NULL, "errorMessage" = NULL, - "updatedAt" = CURRENT_TIMESTAMP + "updatedAt" = (NOW() AT TIME ZONE 'UTC') FROM (VALUES ${updates}) AS updates(id, "channelId", carrier, province, "submitId") WHERE message.id = updates.id AND message.status = 'queued' `); @@ -432,7 +432,7 @@ startSubmitOutboxPublisher() { const values = Prisma.join(failed.map(({ message, reason }) => Prisma.sql`(${message.id}::text, ${reason.slice(0, 1000)}::text)`)); await this.prisma.$executeRaw(Prisma.sql` UPDATE "SmsMessageRecord" AS message - SET status = 'failed', "errorMessage" = failures.reason, "updatedAt" = CURRENT_TIMESTAMP + SET status = 'failed', "errorMessage" = failures.reason, "updatedAt" = (NOW() AT TIME ZONE 'UTC') FROM (VALUES ${values}) AS failures(id, reason) WHERE message.id = failures.id AND message.status = 'queued' `); @@ -728,8 +728,8 @@ startSubmitOutboxPublisher() { SELECT id FROM "GatewaySubmitOutbox" WHERE ( - (status = 'pending' AND "nextAttemptAt" <= CURRENT_TIMESTAMP) - OR (status = 'publishing' AND "leaseExpiresAt" < CURRENT_TIMESTAMP) + (status = 'pending' AND "nextAttemptAt" <= (NOW() AT TIME ZONE 'UTC')) + OR (status = 'publishing' AND "leaseExpiresAt" < (NOW() AT TIME ZONE 'UTC')) ) ORDER BY "createdAt", id LIMIT ${batchSize} @@ -738,9 +738,9 @@ startSubmitOutboxPublisher() { UPDATE "GatewaySubmitOutbox" AS outbox SET status = 'publishing', "leaseOwner" = ${this.submitOutboxLeaseOwner}, - "leaseExpiresAt" = CURRENT_TIMESTAMP + (${leaseSeconds} * INTERVAL '1 second'), + "leaseExpiresAt" = (NOW() AT TIME ZONE 'UTC') + (${leaseSeconds} * INTERVAL '1 second'), "attemptCount" = outbox."attemptCount" + 1, - "updatedAt" = CURRENT_TIMESTAMP + "updatedAt" = (NOW() AT TIME ZONE 'UTC') FROM candidates WHERE outbox.id = candidates.id RETURNING outbox.id, outbox."submitId", outbox.payload @@ -756,11 +756,11 @@ startSubmitOutboxPublisher() { UPDATE "GatewaySubmitOutbox" AS outbox SET status = 'published', "streamEntryId" = published."streamEntryId", - "publishedAt" = CURRENT_TIMESTAMP, + "publishedAt" = (NOW() AT TIME ZONE 'UTC'), "leaseOwner" = NULL, "leaseExpiresAt" = NULL, "lastError" = NULL, - "updatedAt" = CURRENT_TIMESTAMP + "updatedAt" = (NOW() AT TIME ZONE 'UTC') FROM (VALUES ${values}) AS published(id, "streamEntryId") WHERE outbox.id = published.id AND outbox.status = 'publishing' @@ -775,11 +775,11 @@ startSubmitOutboxPublisher() { await this.prisma.$executeRaw(Prisma.sql` UPDATE "GatewaySubmitOutbox" SET status = CASE WHEN "attemptCount" >= 10 THEN 'dead' ELSE 'pending' END, - "nextAttemptAt" = CURRENT_TIMESTAMP + (LEAST(60, POWER(2, LEAST("attemptCount", 6))) * INTERVAL '1 second'), + "nextAttemptAt" = (NOW() AT TIME ZONE 'UTC') + (LEAST(60, POWER(2, LEAST("attemptCount", 6))) * INTERVAL '1 second'), "leaseOwner" = NULL, "leaseExpiresAt" = NULL, "lastError" = ${message.slice(0, 1000)}, - "updatedAt" = CURRENT_TIMESTAMP + "updatedAt" = (NOW() AT TIME ZONE 'UTC') WHERE id = ${row.id} AND "leaseOwner" = ${this.submitOutboxLeaseOwner} `); } catch (recordError) { @@ -1055,7 +1055,7 @@ async refreshTaskProgress(batchTaskId: string, knownSingleMessageStatus?: string WHEN message.status IN ('submit_queued', 'submitted', 'unknown') THEN 'sending' ELSE 'queued' END, - "updatedAt" = CURRENT_TIMESTAMP + "updatedAt" = (NOW() AT TIME ZONE 'UTC') FROM "SmsMessageRecord" AS message WHERE task.id = ${batchTaskId} AND task."sourceType" = 'cmpp' diff --git a/api/src/send-chain/send-inbound-entry.service.ts b/api/src/send-chain/send-inbound-entry.service.ts index 9f405dc..7570a54 100644 --- a/api/src/send-chain/send-inbound-entry.service.ts +++ b/api/src/send-chain/send-inbound-entry.service.ts @@ -28,6 +28,7 @@ type InboundWorkflowPayload = { type ClaimedInboundWorkflow = { id: string; requestKey: string; + tenantId: string; applicationId: string; attempts: number; payload: Prisma.JsonValue; @@ -74,6 +75,7 @@ export class SendInboundEntryService { private readonly logger = new Logger('SendChainService'); private readonly inboundWorkflowWorkerId = `${hostname()}:${process.pid}:${randomUUID()}`; private readonly inboundWorkflowTasks = new Set>(); + private readonly inboundWorkflowActiveTenants = new Set(); private inboundWorkflowTimer?: ReturnType; private inboundWorkflowInFlight = 0; private inboundWorkflowPumping = false; @@ -1241,7 +1243,9 @@ startInboundWorkflowWorker() { this.inboundWorkflowPumping = true; try { await this.refreshInboundWorkflowMetrics(); - const claimed = await this.measureInboundStage('worker_claim', () => this.claimInboundWorkflows(available)); + const activeTenantIds = [...this.inboundWorkflowActiveTenants]; + await this.waitForInboundWorkflowMicroBatch(available, activeTenantIds); + const claimed = await this.measureInboundStage('worker_claim', () => this.claimInboundWorkflows(available, activeTenantIds)); const applications = claimed.length === 0 ? [] : await this.prisma.smsApplication.findMany({ @@ -1250,14 +1254,15 @@ startInboundWorkflowWorker() { }); const applicationById = new Map(applications.map((application) => [application.id, application])); const batchSize = positiveInteger(process.env.API_INBOUND_WORKFLOW_BATCH_SIZE, 64); - for (let offset = 0; offset < claimed.length; offset += batchSize) { - const batch = claimed.slice(offset, offset + batchSize); - this.inboundWorkflowInFlight += batch.length; + for (const [tenantId, tenantItems] of this.groupInboundWorkflowsByTenant(claimed)) { + this.inboundWorkflowActiveTenants.add(tenantId); + this.inboundWorkflowInFlight += tenantItems.length; this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight); - const task = this.processClaimedInboundWorkflowBatch(batch, applicationById) + const task = this.processTenantInboundWorkflowBatches(tenantItems, batchSize, applicationById) .catch((error) => this.logger.error(`CMPP inbound workflow batch failed to settle: ${String(error)}`)) .finally(() => { - this.inboundWorkflowInFlight = Math.max(0, this.inboundWorkflowInFlight - batch.length); + this.inboundWorkflowActiveTenants.delete(tenantId); + this.inboundWorkflowInFlight = Math.max(0, this.inboundWorkflowInFlight - tenantItems.length); this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight); this.inboundWorkflowTasks.delete(task); this.scheduleInboundWorkflowPump(0); @@ -1278,19 +1283,72 @@ startInboundWorkflowWorker() { } } - private claimInboundWorkflows(limit: number) { + private groupInboundWorkflowsByTenant(items: ClaimedInboundWorkflow[]) { + const grouped = new Map(); + for (const item of items) { + const tenantItems = grouped.get(item.tenantId) ?? []; + tenantItems.push(item); + grouped.set(item.tenantId, tenantItems); + } + return grouped; + } + + private async processTenantInboundWorkflowBatches( + items: ClaimedInboundWorkflow[], + batchSize: number, + applicationById: Map, + ) { + // A tenant account is one financial consistency boundary. Chunks for the + // same tenant stay serial, while different tenants are started as separate + // tasks by the pump and can use the configured workflow concurrency. + for (let offset = 0; offset < items.length; offset += batchSize) { + await this.processClaimedInboundWorkflowBatch(items.slice(offset, offset + batchSize), applicationById); + } + } + + private async waitForInboundWorkflowMicroBatch(limit: number, excludedTenantIds: string[]) { + const maxWaitMs = Math.min(250, getNonNegativeConfigInteger(process.env, 'API_INBOUND_WORKFLOW_BATCH_WAIT_MS', 40)); + if (maxWaitMs === 0 || limit < 2) return; + const configuredBatchSize = positiveInteger(process.env.API_INBOUND_WORKFLOW_BATCH_SIZE, 64); + const targetBatchSize = Math.min( + limit, + configuredBatchSize, + positiveInteger(process.env.API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE, 32), + ); + if (targetBatchSize < 2) return; + const excludedFilter = excludedTenantIds.length === 0 + ? Prisma.empty + : Prisma.sql`AND "tenantId" NOT IN (${Prisma.join(excludedTenantIds)})`; + const rows = await this.prisma.$queryRaw>(Prisma.sql` + SELECT COUNT(*)::bigint AS count + FROM "CmppInboundSubmissionInbox" + WHERE status = 'pending' + AND "nextAttemptAt" <= (NOW() AT TIME ZONE 'UTC') + ${excludedFilter} + `); + const ready = Number(rows[0]?.count ?? 0); + if (ready > 0 && ready < targetBatchSize) await sleep(maxWaitMs); + } + + private claimInboundWorkflows(limit: number, excludedTenantIds: string[] = []) { const staleSeconds = positiveInteger(process.env.API_INBOUND_WORKFLOW_STALE_SECONDS, 300); + const excludedFilter = excludedTenantIds.length === 0 + ? Prisma.empty + : Prisma.sql`AND "tenantId" NOT IN (${Prisma.join(excludedTenantIds)})`; return this.prisma.$queryRaw(Prisma.sql` WITH candidates AS ( SELECT id FROM "CmppInboundSubmissionInbox" WHERE ( - status = 'pending' - AND "nextAttemptAt" <= (NOW() AT TIME ZONE 'UTC') - ) OR ( - status = 'processing' - AND "lockedAt" <= (NOW() AT TIME ZONE 'UTC') - make_interval(secs => ${staleSeconds}) + ( + status = 'pending' + AND "nextAttemptAt" <= (NOW() AT TIME ZONE 'UTC') + ) OR ( + status = 'processing' + AND "lockedAt" <= (NOW() AT TIME ZONE 'UTC') - make_interval(secs => ${staleSeconds}) + ) ) + ${excludedFilter} -- Priority applications enter the same durable Inbox, but are claimed first while -- preserving FIFO within each class. This keeps the V5 priority contract effective -- before BullMQ without adding another non-durable queue. @@ -1306,7 +1364,7 @@ startInboundWorkflowWorker() { "updatedAt" = (NOW() AT TIME ZONE 'UTC') FROM candidates WHERE inbox.id = candidates.id - RETURNING inbox.id, inbox."requestKey", inbox."applicationId", inbox.attempts, inbox.payload + RETURNING inbox.id, inbox."requestKey", inbox."tenantId", inbox."applicationId", inbox.attempts, inbox.payload `); } diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index ff394a1..b03561e 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -2159,3 +2159,11 @@ - Gateway拉取待投递回执必须按账号进程内单飞，API必须用FOR UPDATE SKIP LOCKED将pending原子领取为带租约的dispatching；未实际发送的领取必须无损释放，租约过期可恢复。SubmitResp后只能合并调度账号级刷新，不得每条并发扫描同一批pending记录。 - API新建回执后的直推与Gateway恢复拉取必须共享同一dispatching + claimId所有权；直推未抢到记录时必须退出，不得与恢复路径各发一次。ACK超时定时器必须在ACK注册锁内完成指针赋值，避免极短截止时间下的竞态。 - 2026-08-25修复后正价阶梯客户入口20/30/50/70/100/150/200 TPS均零拒绝、零节流、零连接错误；完整供应商首提在150/200冲击档分别约94.86/92.24 TPS，显示端到端容量天花板约95 TPS。二次直推领取修复后100 TPS正价复验为999/999、P95/P99=102/179ms、999次999个唯一首提在12.488秒完成（80.00 TPS），972条已形成终态回执的下游投递生命周期尝试次数恰为972、重复0、最大1。生产建议仍保留容量余量，建议限速70 TPS，不将客户SubmitResp受理200 TPS误作完整供应商TPS。 + +## 单 Gateway 单企业工作流微批治理（2026-08-25） + +- 入站工作流按企业分组领取和执行：不同企业可以并行使用工作流槽位，同一企业的多个批次必须串行，保持账户冻结、日限额和频控的一致性边界；当前阶段不得通过增加Gateway实例规避该边界。 +- Worker领取前允许最多`API_INBOUND_WORKFLOW_BATCH_WAIT_MS`毫秒的有界聚合等待，默认40、合法范围0～250；目标批量由`API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE`控制，默认32且不得超过`API_INBOUND_WORKFLOW_BATCH_SIZE`。等待仅在已有少量就绪记录时发生，空队列不得固定休眠。 +- 正在处理的企业必须从下一次领取候选中排除，避免同企业并发事务重新争抢账户锁；租约过期恢复、优先级/FIFO、`FOR UPDATE SKIP LOCKED`和逐请求幂等语义保持不变。 +- 原生SQL对Prisma的无时区时间列统一使用`NOW() AT TIME ZONE 'UTC'`，覆盖Submit Outbox领取、租约、发布、重试及关联消息更新时间，禁止用数据库会话时区污染时延审计。 +- 验收必须保持正价，分别执行单企业100/150 TPS和至少两个独立企业合计200 TPS，按非补发首次供应商Submit、回执、上行、计费、主备补发、业务拦截及全队列排空对账；触发拒绝、连接错误、持续积压、数据库异常或账务不一致立即停止。多Gateway P2不在本阶段范围。 diff --git a/docs/phase-5-gateway-capacity-expansion-plan.md b/docs/phase-5-gateway-capacity-expansion-plan.md index 726953b..9866c32 100644 --- a/docs/phase-5-gateway-capacity-expansion-plan.md +++ b/docs/phase-5-gateway-capacity-expansion-plan.md @@ -507,3 +507,9 @@ API 必须返回逐事件结果，而不是只有整批成功或失败： 5. 只有单Gateway资源被实测打满后才启动多Gateway P2。 诊断还发现Outbox原生SQL使用`CURRENT_TIMESTAMP`，与Prisma UTC时间混用时会产生8小时审计偏差。该问题不影响本次吞吐和队列排空结论，但必须在下一次代码改造中统一为UTC并回归租约、重试和发布时延。 + +## 15. 单企业工作流微批治理实施范围 + +本轮只在单Gateway架构内处理95 TPS定向诊断暴露的工作流问题，不实施多Gateway P2。具体改造为：领取前最多等待40ms把小批合并到目标32条；按企业分组，不同企业并行、同企业各批串行；领取SQL排除正在执行的企业；Submit Outbox及关联消息的原生SQL时间统一为UTC。环境参数必须显式配置并由发布脚本校验，目标批量不得超过实际批次上限。 + +发布验收保持应用单价325以及现有临时号段规则和白名单。先做单企业100/150 TPS，以确认原约95 TPS边界是否改善；再使用至少两个独立企业做聚合200 TPS，用于区分单企业账户一致性边界和平台总吞吐。所有档位都按非补发首次供应商Submit计算，并对账Submit、回执、上行、账务、主备补发、拦截规则与队列排空；停止条件沿用第五阶段，不以入口SubmitResp替代完整链路结论。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index 6ae7011..2b3ea60 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -4881,3 +4881,7 @@ npm run verify:phase8 | TC-CMPP-PHASE5-010 | API直推与Gateway恢复同时命中同一回执 | 仅抢到dispatching + claimId的路径发送；100 TPS正价实测972条终态回执对应972次下游ACK，重复0，最大尝试1 | | TC-CMPP-PHASE5-011 | claim未发送、租约过期和ACK定时器竞态 | SubmitResp屏障或客户离线时释放claim且不增加重试次数；过期dispatching可恢复；Linux go test -race ./internal/inbound通过 | | TC-CMPP-PHASE5-012 | 修复后正价容量与口径分离 | 20至200 TPS入口均无拒绝、节流或连接错误；999条100 TPS复验入口P95/P99 102/179ms，供应商首提80.00 TPS；150/200冲击档首提约95 TPS天花板，两种口径分开报告 | +| TC-CMPP-PHASE5-013 | 同企业工作流串行与有界微批 | 同企业领取期间不再领取第二批；已有少量记录最多等待40ms聚合，目标32、上限64；优先级/FIFO、租约恢复和幂等不变 | +| TC-CMPP-PHASE5-014 | 不同企业正价并行 | 至少两个独立企业合计200 TPS；各企业账户冻结串行且企业间并行，单价均为325，分别对账消息、账单、首提和最终状态 | +| TC-CMPP-PHASE5-015 | Outbox UTC时间语义 | Asia/Shanghai数据库会话下领取、租约、发布、重试均写UTC无时区值；publishedAt-createdAt不再出现约8小时偏差 | +| TC-CMPP-PHASE5-016 | 微批发布边界与停止线 | 仅单Gateway；分别验证单企业100/150与多企业200，发生拒绝、连接错误、持续积压、数据库异常或账务不一致立即停止 | diff --git a/tools/deploy/production-bootstrap.sh b/tools/deploy/production-bootstrap.sh index 33e8cad..11ccbac 100644 --- a/tools/deploy/production-bootstrap.sh +++ b/tools/deploy/production-bootstrap.sh @@ -18,6 +18,8 @@ CMPP_INBOUND_WORKFLOW_WORKER_ENABLED="${CMPP_INBOUND_WORKFLOW_WORKER_ENABLED:-tr API_INBOUND_WORKFLOW_CONCURRENCY="${API_INBOUND_WORKFLOW_CONCURRENCY:-32}" API_INBOUND_WORKFLOW_BATCH_ENABLED="${API_INBOUND_WORKFLOW_BATCH_ENABLED:-true}" API_INBOUND_WORKFLOW_BATCH_SIZE="${API_INBOUND_WORKFLOW_BATCH_SIZE:-64}" +API_INBOUND_WORKFLOW_BATCH_WAIT_MS="${API_INBOUND_WORKFLOW_BATCH_WAIT_MS:-40}" +API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE="${API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE:-32}" API_INBOUND_WORKFLOW_POLL_INTERVAL_MS="${API_INBOUND_WORKFLOW_POLL_INTERVAL_MS:-100}" API_INBOUND_WORKFLOW_STALE_SECONDS="${API_INBOUND_WORKFLOW_STALE_SECONDS:-300}" API_DB_POOL_MAX="${API_DB_POOL_MAX:-32}" @@ -209,6 +211,8 @@ CMPP_INBOUND_WORKFLOW_WORKER_ENABLED=${CMPP_INBOUND_WORKFLOW_WORKER_ENABLED} API_INBOUND_WORKFLOW_CONCURRENCY=${API_INBOUND_WORKFLOW_CONCURRENCY} API_INBOUND_WORKFLOW_BATCH_ENABLED=${API_INBOUND_WORKFLOW_BATCH_ENABLED} API_INBOUND_WORKFLOW_BATCH_SIZE=${API_INBOUND_WORKFLOW_BATCH_SIZE} +API_INBOUND_WORKFLOW_BATCH_WAIT_MS=${API_INBOUND_WORKFLOW_BATCH_WAIT_MS} +API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE=${API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE} API_INBOUND_WORKFLOW_POLL_INTERVAL_MS=${API_INBOUND_WORKFLOW_POLL_INTERVAL_MS} API_INBOUND_WORKFLOW_STALE_SECONDS=${API_INBOUND_WORKFLOW_STALE_SECONDS} API_DB_POOL_MAX=${API_DB_POOL_MAX} diff --git a/tools/deploy/production-deploy.sh b/tools/deploy/production-deploy.sh index cb043c9..23d14e5 100644 --- a/tools/deploy/production-deploy.sh +++ b/tools/deploy/production-deploy.sh @@ -42,6 +42,14 @@ if [[ "${API_INBOUND_WORKFLOW_BATCH_ENABLED:-}" != "true" || ! "${API_INBOUND_WO echo "API_INBOUND_WORKFLOW_BATCH_ENABLED=true and a positive API_INBOUND_WORKFLOW_BATCH_SIZE are required in $ENV_FILE." >&2 exit 1 fi +if [[ ! "${API_INBOUND_WORKFLOW_BATCH_WAIT_MS:-}" =~ ^[0-9]+$ || "${API_INBOUND_WORKFLOW_BATCH_WAIT_MS}" -gt 250 ]]; then + echo "API_INBOUND_WORKFLOW_BATCH_WAIT_MS must be between 0 and 250 in $ENV_FILE." >&2 + exit 1 +fi +if [[ ! "${API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE:-}" =~ ^[1-9][0-9]*$ || "${API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE}" -gt "${API_INBOUND_WORKFLOW_BATCH_SIZE}" ]]; then + echo "API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE must be positive and no greater than API_INBOUND_WORKFLOW_BATCH_SIZE in $ENV_FILE." >&2 + exit 1 +fi if [[ ! "${API_DB_POOL_MAX:-}" =~ ^[1-9][0-9]*$ || ! "${API_WORKER_DB_POOL_MAX:-}" =~ ^[1-9][0-9]*$ ]]; then echo "API_DB_POOL_MAX and API_WORKER_DB_POOL_MAX must be positive integers in $ENV_FILE." >&2 exit 1 diff --git a/tools/deploy/verify-production-deployment.mjs b/tools/deploy/verify-production-deployment.mjs index bfce981..4474dc3 100644 --- a/tools/deploy/verify-production-deployment.mjs +++ b/tools/deploy/verify-production-deployment.mjs @@ -33,6 +33,8 @@ for (const marker of [ 'API_INBOUND_WORKFLOW_CONCURRENCY', 'API_INBOUND_WORKFLOW_BATCH_ENABLED=true', 'API_INBOUND_WORKFLOW_BATCH_SIZE', + 'API_INBOUND_WORKFLOW_BATCH_WAIT_MS', + 'API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE', 'cmpp-send-worker.service', 'Environment=CMPP_PROCESS_ROLE=api', 'Environment=CMPP_PROCESS_ROLE=worker',