From 52028b9bbdea20f0cd02efe792e91b558ba50be9 Mon Sep 17 00:00:00 2001 From: hectorzhao Date: Fri, 21 Aug 2026 09:44:51 +0800 Subject: [PATCH] perf: batch CMPP inbound workflow processing --- .env.example | 2 + api/src/metrics/metrics.service.ts | 3 + .../phone-frequency.service.spec.ts | 31 ++ .../risk-review/phone-frequency.service.ts | 144 +++++++ .../risk-review/risk-review.service.spec.ts | 18 + api/src/risk-review/risk-review.service.ts | 75 ++++ .../send-chain/send-inbound-entry.service.ts | 355 +++++++++++++++++- docs/codebase-modularization-roadmap.md | 2 + .../first-version-development-requirements.md | 7 + docs/production-deployment.md | 3 +- docs/system-functional-test-cases.md | 12 + docs/testing-progress.md | 7 + tools/deploy/production-bootstrap.sh | 4 + tools/deploy/production-deploy.sh | 4 + tools/deploy/verify-production-deployment.mjs | 2 + 15 files changed, 662 insertions(+), 7 deletions(-) diff --git a/.env.example b/.env.example index 88098ec..0358eee 100644 --- a/.env.example +++ b/.env.example @@ -18,6 +18,8 @@ API_WORKER_METRICS_PORT=9465 CMPP_INBOUND_FAST_PATH_ENABLED=true CMPP_INBOUND_WORKFLOW_WORKER_ENABLED=true API_INBOUND_WORKFLOW_CONCURRENCY=32 +API_INBOUND_WORKFLOW_BATCH_ENABLED=true +API_INBOUND_WORKFLOW_BATCH_SIZE=64 API_INBOUND_WORKFLOW_POLL_INTERVAL_MS=100 API_INBOUND_WORKFLOW_STALE_SECONDS=300 ADMIN_SESSION_IDLE_TIMEOUT_MS=3600000 diff --git a/api/src/metrics/metrics.service.ts b/api/src/metrics/metrics.service.ts index c67a050..61b320c 100644 --- a/api/src/metrics/metrics.service.ts +++ b/api/src/metrics/metrics.service.ts @@ -7,6 +7,9 @@ const CMPP_INBOUND_DURATION_BUCKETS = [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, export type CmppInboundStage = | 'application_lookup' | 'inbox_persist' + | 'worker_claim' + | 'reference_preload' + | 'daily_quota' | 'long_message_fragment' | 'submission_precheck' | 'template_match' diff --git a/api/src/risk-review/phone-frequency.service.spec.ts b/api/src/risk-review/phone-frequency.service.spec.ts index 657cc9b..eee701f 100644 --- a/api/src/risk-review/phone-frequency.service.spec.ts +++ b/api/src/risk-review/phone-frequency.service.spec.ts @@ -1,6 +1,37 @@ import { PhoneFrequencyService, fixedShanghaiWindow } from './phone-frequency.service'; describe('PhoneFrequencyService', () => { + it('persists independent idempotency results for a unique-phone batch in one transaction', async () => { + const tx = { + phoneFrequencyReservation: { + findMany: jest.fn().mockResolvedValue([]), + createMany: jest.fn().mockResolvedValue({ count: 2 }), + }, + phoneFrequencyWhitelist: { findMany: jest.fn().mockResolvedValue([]) }, + }; + const prisma = { + riskRule: { findMany: jest.fn().mockResolvedValue([]) }, + $transaction: jest.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), + }; + const riskReview = { ensureDefaultRules: jest.fn().mockResolvedValue(undefined) }; + const service = new PhoneFrequencyService(prisma as never, riskReview as never); + + const results = await service.reserveBatch([ + { tenantId: 'tenant-1', applicationId: 'app-1', phoneNumber: '13800000001', reservationKey: 'inbox-1:frequency' }, + { tenantId: 'tenant-1', applicationId: 'app-1', phoneNumber: '13800000002', reservationKey: 'inbox-2:frequency' }, + ]); + + expect(results.get('inbox-1:frequency')?.size).toBe(0); + expect(results.get('inbox-2:frequency')?.size).toBe(0); + expect(prisma.$transaction).toHaveBeenCalledTimes(1); + expect(tx.phoneFrequencyReservation.createMany).toHaveBeenCalledWith({ + data: expect.arrayContaining([ + expect.objectContaining({ reservationKey: 'inbox-1:frequency', result: [] }), + expect.objectContaining({ reservationKey: 'inbox-2:frequency', result: [] }), + ]), + }); + }); + it('aligns five-minute cycles and natural days in Asia/Shanghai', () => { const requestedAt = new Date('2026-07-30T16:07:42.000Z'); diff --git a/api/src/risk-review/phone-frequency.service.ts b/api/src/risk-review/phone-frequency.service.ts index 06fb866..67e9a47 100644 --- a/api/src/risk-review/phone-frequency.service.ts +++ b/api/src/risk-review/phone-frequency.service.ts @@ -63,6 +63,15 @@ export interface PhoneFrequencyRejection { reason: string; } +export interface PhoneFrequencyBatchReservation { + tenantId: string; + applicationId: string; + phoneNumber: string; + reservationKey: string; + sourceType?: string; + requestedAt?: Date; +} + @Injectable() export class PhoneFrequencyService { constructor( @@ -180,6 +189,141 @@ export class PhoneFrequencyService { }, { isolationLevel: Prisma.TransactionIsolationLevel.ReadCommitted }); } + /** + * Reserve independent one-phone Inbox items in bounded database batches. The + * reservation rows and counters commit in the same transaction, so reclaiming + * any subset replays its original decision. Duplicate phones intentionally use + * the established single-item path because their within-batch threshold order + * is business-significant. + */ + async reserveBatch(items: PhoneFrequencyBatchReservation[]) { + const results = new Map>(); + if (items.length === 0) return results; + const reservationKeys = items.map((item) => item.reservationKey.trim()); + if (reservationKeys.some((key) => !key) || new Set(reservationKeys).size !== reservationKeys.length) { + throw new BadRequestException('号码频控批次幂等键为空或重复'); + } + const groups = new Map(); + for (const item of items) { + const key = `${item.tenantId}:${item.applicationId}`; + const group = groups.get(key) ?? []; + group.push({ ...item, phoneNumber: item.phoneNumber.trim(), reservationKey: item.reservationKey.trim() }); + groups.set(key, group); + } + for (const group of groups.values()) { + const uniquePhones = new Set(group.map((item) => item.phoneNumber)); + if (uniquePhones.size !== group.length) { + for (const item of group) { + results.set(item.reservationKey, await this.reserve( + item.tenantId, + item.applicationId, + [item.phoneNumber], + item.sourceType, + item.requestedAt ?? new Date(), + item.reservationKey, + )); + } + continue; + } + await this.riskReview.ensureDefaultRules(); + const rules = await this.effectiveRules(group[0].applicationId); + const groupResults = await this.prisma.$transaction(async (tx) => { + const output = new Map>(); + const existing = await tx.phoneFrequencyReservation.findMany({ + where: { reservationKey: { in: group.map((item) => item.reservationKey) } }, + }); + const existingByKey = new Map(existing.map((item) => [item.reservationKey, item])); + const missing: PhoneFrequencyBatchReservation[] = []; + for (const item of group) { + const replay = existingByKey.get(item.reservationKey); + if (!replay) { + missing.push(item); + continue; + } + if (replay.tenantId !== item.tenantId || replay.applicationId !== item.applicationId) { + throw new BadRequestException('号码频控幂等键已用于另一笔预留'); + } + output.set(item.reservationKey, frequencyRejectionsFromJson(replay.result)); + } + if (missing.length === 0) return output; + const whitelistedPhones = await this.findActiveWhitelistedPhones(tx, missing.map((item) => item.phoneNumber)); + const controlled = missing.filter((item) => !whitelistedPhones.has(item.phoneNumber)); + const rejectedByPhone = new Map(); + for (const rule of rules) { + const byWindow = new Map(); + for (const item of controlled) { + const window = fixedShanghaiWindow(item.requestedAt ?? new Date(), readPeriodSeconds(rule)); + const key = `${window.startAt.toISOString()}:${window.endAt.toISOString()}`; + const bucket = byWindow.get(key) ?? { ...window, items: [] }; + bucket.items.push(item); + byWindow.set(key, bucket); + } + for (const bucket of byWindow.values()) { + const states = await this.upsertStates(tx, { + tenantId: group[0].tenantId, + applicationId: group[0].applicationId, + phones: bucket.items.map((item) => item.phoneNumber), + rule, + window: { startAt: bucket.startAt, endAt: bucket.endAt }, + }); + const newTriggers = states.filter((state) => state.activeHitId === null && state.count > rule.thresholdValue); + const hitByStateId = new Map(); + if (newTriggers.length > 0) { + await tx.phoneFrequencyHit.createMany({ + data: newTriggers.map((state) => { + const hitId = randomUUID(); + hitByStateId.set(state.id, hitId); + return { + id: hitId, + tenantId: group[0].tenantId, + applicationId: group[0].applicationId, + ruleId: rule.id, + ruleCode: rule.code, + ruleName: rule.name, + phoneNumber: state.phoneNumber, + thresholdValue: Math.floor(rule.thresholdValue), + actualValue: state.count, + windowStartedAt: state.windowStartedAt, + windowEndsAt: state.windowEndsAt, + generation: state.generation, + action: 'block', + sourceType: bucket.items[0]?.sourceType, + }; + }), + }); + await this.attachActiveHits(tx, hitByStateId); + } + for (const state of states) { + if (state.activeHitId === null && state.count <= rule.thresholdValue) continue; + const reason = `${rule.name}命中：本周期最多${Math.floor(rule.thresholdValue)}条，当前第${state.count}条，周期${formatWindow(state.windowStartedAt, state.windowEndsAt)}`; + const previous = rejectedByPhone.get(state.phoneNumber); + rejectedByPhone.set(state.phoneNumber, { + code: 'PHONE_FREQUENCY_LIMIT', + reason: previous ? `${previous.reason}；${reason}` : reason, + }); + } + } + } + await tx.phoneFrequencyReservation.createMany({ + data: missing.map((item) => { + const rejection = rejectedByPhone.get(item.phoneNumber); + const result = rejection ? [{ phoneNumber: item.phoneNumber, ...rejection }] : []; + output.set(item.reservationKey, frequencyRejectionsFromJson(result)); + return { + reservationKey: item.reservationKey, + tenantId: item.tenantId, + applicationId: item.applicationId, + result, + }; + }), + }); + return output; + }, { isolationLevel: Prisma.TransactionIsolationLevel.ReadCommitted }); + for (const [key, value] of groupResults) results.set(key, value); + } + return results; + } + async listHits(query: PhoneFrequencyHitQuery) { const page = Math.max(1, Math.floor(Number(query.page) || 1)); const pageSize = Math.min(100, Math.max(1, Math.floor(Number(query.pageSize) || 20))); diff --git a/api/src/risk-review/risk-review.service.spec.ts b/api/src/risk-review/risk-review.service.spec.ts index 4efc346..2594605 100644 --- a/api/src/risk-review/risk-review.service.spec.ts +++ b/api/src/risk-review/risk-review.service.spec.ts @@ -27,6 +27,7 @@ function createPrismaMock(overrides: Record = {}) { }, smsTemplate: { findUnique: jest.fn().mockResolvedValue(null), + findMany: jest.fn().mockResolvedValue([]), }, smsSendTask: { create: jest.fn().mockImplementation(({ data }: { data: Record }) => @@ -62,6 +63,23 @@ function createPrismaMock(overrides: Record = {}) { } describe('RiskReviewService', () => { + it('shares read-only rule inputs across an approved CMPP evaluation batch', async () => { + const prisma = createPrismaMock(); + const service = new RiskReviewService(prisma as never); + + const results = await service.evaluateTasksBatch([ + { tenantId: 'tenant-1', applicationId: 'app-1', content: '【测试】验证码000001', phones: ['13800000001'], sourceType: 'cmpp' }, + { tenantId: 'tenant-1', applicationId: 'app-1', content: '【测试】验证码000002', phones: ['13800000002'], sourceType: 'cmpp' }, + ]); + + expect(results).toHaveLength(2); + expect(results.every((result) => result.status === 'approved')).toBe(true); + expect(prisma.riskRule.findMany).toHaveBeenCalledTimes(1); + expect(prisma.sensitiveWord.findMany).toHaveBeenCalledTimes(1); + expect(prisma.smsBatchTask.count).not.toHaveBeenCalled(); + expect(prisma.smsSendTask.create).not.toHaveBeenCalled(); + }); + it('coalesces concurrent default-rule checks and reuses the short completeness cache', async () => { const prisma = createPrismaMock(); let releaseCount: ((count: number) => void) | undefined; diff --git a/api/src/risk-review/risk-review.service.ts b/api/src/risk-review/risk-review.service.ts index b6b015b..fde6c4b 100644 --- a/api/src/risk-review/risk-review.service.ts +++ b/api/src/risk-review/risk-review.service.ts @@ -433,6 +433,81 @@ export class RiskReviewService { }; } + /** + * Evaluate an Inbox claim batch against one database snapshot of the read-only + * rule inputs. Approved items do not need their own template/rule/sensitive-word + * queries; exceptional decisions still go through evaluateTask so their audit + * task and hit rows keep the existing semantics. + */ + async evaluateTasksBatch(items: EvaluateSmsTaskDto[]) { + if (items.length === 0) return []; + await this.ensureDefaultRules(); + if (items.some((item) => item.createdById)) { + // The CMPP worker never supplies createdById. Keep the general API honest + // instead of silently weakening its foreign-key validation in the fast path. + return Promise.all(items.map((item) => this.evaluateTask(item))); + } + const applicationIds = [...new Set(items.map((item) => item.applicationId).filter((id): id is string => Boolean(id)))]; + const templateIds = [...new Set(items.map((item) => item.templateId).filter((id): id is string => Boolean(id)))]; + const [templates, sensitiveWords, applicationInputs] = await Promise.all([ + templateIds.length + ? this.prisma.smsTemplate.findMany({ where: { id: { in: templateIds } }, include: { variables: true } }) + : Promise.resolve([]), + this.prisma.sensitiveWord.findMany({ where: { status: 'active' }, select: { word: true, level: true } }), + Promise.all(applicationIds.map(async (applicationId) => ({ + applicationId, + rules: await this.effectiveRules(applicationId), + recentTaskCount: await this.countRecentClientTasks(applicationId, 'cmpp'), + }))), + ]); + const templateById = new Map(templates.map((template) => [template.id, template])); + const inputsByApplication = new Map(applicationInputs.map((entry) => [entry.applicationId, entry])); + + return Promise.all(items.map(async (data) => { + const phones = data.phones ?? []; + const uniquePhones = [...new Set(phones)]; + const phoneTotal = phones.length; + const template = data.templateId ? templateById.get(data.templateId) : undefined; + const variableIssues = evaluateTemplateVariables(template?.variables ?? [], data.content, data.variables ?? {}); + const contentIssues = evaluateContent(data.content, sensitiveWords); + if (variableIssues.length > 0) { + contentIssues.push({ + ruleCode: 'TEMPLATE_VARIABLE_INVALID', + ruleName: '模板变量校验失败', + thresholdValue: 0, + actualValue: variableIssues.length, + action: 'block', + reason: formatTemplateVariableIssueReason(variableIssues), + }); + } + const applicationInput = data.applicationId ? inputsByApplication.get(data.applicationId) : undefined; + const rules = applicationInput?.rules ?? []; + const requestedAt = data.requestedAt ? new Date(data.requestedAt) : new Date(); + const nonWorkingRule = rules.find((rule) => rule.code === 'NON_WORKING_MARKETING_BULK'); + const nonWorkingMarketingPhones = isMarketing(data.category ?? template?.category) + && isNonWorkingTime(requestedAt, readNonWorkingConfig(nonWorkingRule?.config)) + ? phoneTotal + : 0; + const hits = this.evaluateRules(rules, { + phoneTotal, + nonWorkingMarketingPhones, + recentTaskCount: applicationInput?.recentTaskCount ?? 0, + }); + hits.push(...contentIssues.map(contentIssueToHit)); + const decision = decideRiskAction(hits); + if (decision.status !== 'approved') { + return this.evaluateTask(data); + } + return { + canSubmit: true, + status: decision.status, + riskDecision: decision.riskDecision, + reason: hits.length > 0 ? hits.map((hit) => hit.reason).join('; ') : null, + task: null, + }; + })); + } + async approveTask(taskId: string, data: ReviewSmsTaskDto) { const task = await this.prisma.smsSendTask.findUnique({ where: { id: taskId } }); if (!task) { diff --git a/api/src/send-chain/send-inbound-entry.service.ts b/api/src/send-chain/send-inbound-entry.service.ts index c0a80e2..ec86441 100644 --- a/api/src/send-chain/send-inbound-entry.service.ts +++ b/api/src/send-chain/send-inbound-entry.service.ts @@ -41,6 +41,20 @@ type PersistedInboundWorkflowRow = { type InboundApplication = NonNullable>>; +type InboundTemplate = Prisma.SmsTemplateGetPayload<{ include: { signature: true } }>; + +type InboundBatchCandidate = { + item: ClaimedInboundWorkflow; + payload: InboundWorkflowPayload; + application: InboundApplication; + phoneNumber: string; + template?: InboundTemplate; + signatureId: string; + templateVariables?: Record; + drainageInfoId?: string; + clientSrcId: string | null; +}; + type InboundWorkflowResponse = { accepted: boolean; tenantId: string; @@ -1227,7 +1241,7 @@ startInboundWorkflowWorker() { this.inboundWorkflowPumping = true; try { await this.refreshInboundWorkflowMetrics(); - const claimed = await this.claimInboundWorkflows(available); + const claimed = await this.measureInboundStage('worker_claim', () => this.claimInboundWorkflows(available)); const applications = claimed.length === 0 ? [] : await this.prisma.smsApplication.findMany({ @@ -1235,13 +1249,15 @@ startInboundWorkflowWorker() { include: { tenant: true, ipAllowlist: true }, }); const applicationById = new Map(applications.map((application) => [application.id, application])); - for (const item of claimed) { - this.inboundWorkflowInFlight += 1; + 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; this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight); - const task = this.processClaimedInboundWorkflow(item, applicationById.get(item.applicationId)) - .catch((error) => this.logger.error(`CMPP inbound workflow ${item.id} failed to settle: ${String(error)}`)) + const task = this.processClaimedInboundWorkflowBatch(batch, applicationById) + .catch((error) => this.logger.error(`CMPP inbound workflow batch failed to settle: ${String(error)}`)) .finally(() => { - this.inboundWorkflowInFlight = Math.max(0, this.inboundWorkflowInFlight - 1); + this.inboundWorkflowInFlight = Math.max(0, this.inboundWorkflowInFlight - batch.length); this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight); this.inboundWorkflowTasks.delete(task); this.scheduleInboundWorkflowPump(0); @@ -1294,6 +1310,333 @@ startInboundWorkflowWorker() { `); } + private async processClaimedInboundWorkflowBatch( + items: ClaimedInboundWorkflow[], + applicationById: Map, + ) { + if (process.env.API_INBOUND_WORKFLOW_BATCH_ENABLED === 'false' || items.length < 2) { + await Promise.all(items.map((item) => this.processClaimedInboundWorkflow(item, applicationById.get(item.applicationId)))); + return; + } + let completed = new Set(); + try { + completed = await this.processCommonInboundWorkflowBatch(items, applicationById); + } catch (error) { + // The common path is wholly idempotent: quota/frequency reservations have + // stable keys, records have stable message/task keys, and BullMQ job IDs are + // message IDs. Falling back after a partial external failure repairs rather + // than duplicates the individual items. + this.logger.warn(`CMPP inbound common batch fell back to individual recovery: ${String(error)}`); + } + await Promise.all(items + .filter((item) => !completed.has(item.id)) + .map((item) => this.processClaimedInboundWorkflow(item, applicationById.get(item.applicationId)))); + } + + private async processCommonInboundWorkflowBatch( + items: ClaimedInboundWorkflow[], + applicationById: Map, + ) { + const parsed = items.flatMap((item) => { + try { + const payload = parseInboundWorkflowPayload(item.payload); + const application = applicationById.get(item.applicationId); + return payload.phoneNumbers.length === 1 && application ? [{ item, payload, application }] : []; + } catch { + return []; + } + }); + if (parsed.length < 2) return new Set(); + const applicationIds = [...new Set(parsed.map((entry) => entry.application.id))]; + const phones = [...new Set(parsed.map((entry) => entry.payload.phoneNumbers[0]))]; + const messageIds = parsed.flatMap((entry) => entry.payload.messageIds); + const tenantIds = [...new Set(parsed.map((entry) => entry.application.tenantId))]; + const [templates, signatures, globalBlacklist, enterpriseBlacklist, persisted] = await this.measureInboundStage('reference_preload', () => Promise.all([ + this.prisma.smsTemplate.findMany({ + where: { applicationId: { in: applicationIds }, auditStatus: 'approved', signature: { auditStatus: 'approved' } }, + include: { signature: true }, + orderBy: { updatedAt: 'desc' }, + }), + this.prisma.smsSignature.findMany({ + where: { applicationId: { in: applicationIds }, auditStatus: 'approved' }, + select: { id: true, applicationId: true, name: true, updatedAt: true }, + orderBy: { updatedAt: 'desc' }, + }), + this.prisma.globalBlacklist.findMany({ + where: { phoneNumber: { in: phones }, status: 'active' }, + select: { phoneNumber: true }, + }), + this.prisma.enterpriseBlacklist.findMany({ + where: { tenantId: { in: tenantIds }, applicationId: { in: applicationIds }, phoneNumber: { in: phones }, status: 'active' }, + select: { tenantId: true, applicationId: true, phoneNumber: true }, + }), + this.prisma.smsMessageRecord.findMany({ where: { messageId: { in: messageIds } }, select: { messageId: true } }), + ])); + const globalRejected = new Set(globalBlacklist.map((entry) => entry.phoneNumber)); + const enterpriseRejected = new Set(enterpriseBlacklist.map((entry) => `${entry.tenantId}:${entry.applicationId}:${entry.phoneNumber}`)); + const persistedIds = new Set(persisted.map((entry) => entry.messageId)); + const templatesByApplication = new Map(); + for (const template of templates) { + const values = templatesByApplication.get(template.applicationId) ?? []; + values.push(template); + templatesByApplication.set(template.applicationId, values); + } + const signaturesByApplication = new Map(); + for (const signature of signatures) { + if (!signature.applicationId) continue; + const values = signaturesByApplication.get(signature.applicationId) ?? []; + values.push(signature); + signaturesByApplication.set(signature.applicationId, values); + } + const candidates: InboundBatchCandidate[] = []; + for (const entry of parsed) { + const { item, payload, application } = entry; + const data = payload.data; + const phoneNumber = payload.phoneNumbers[0]; + if (persistedIds.has(payload.messageIds[0]) + || application.cmppAccount !== data.account + || application.status !== 'active' + || application.tenant.status !== 'active' + || !application.interfaceEnabled + || application.tenant.certificationStatus !== 'approved' + || !/^1\d{10}$/.test(phoneNumber) + || globalRejected.has(phoneNumber) + || enterpriseRejected.has(`${application.tenantId}:${application.id}:${phoneNumber}`) + // Paid messages retain the existing per-message account lock until the + // dedicated batch-ledger migration is introduced; never weaken billing + // correctness merely to increase the benchmark number. + || moneyToNumber(application.customerUnitPrice) !== 0) continue; + try { + if (data.remoteIp && !isIpAllowed(data.remoteIp, application.ipAllowlist.map((allowlist) => allowlist.ipCidr))) continue; + const clientSrcId = validateInboundApplicationSrcId(data.srcId, application); + const template = (templatesByApplication.get(application.id) ?? []).find((candidate) => ( + candidate.content === data.content || matchTemplateContent(candidate.content, data.content) !== null + )); + const templateVariables = template ? matchTemplateContent(template.content, data.content) ?? {} : undefined; + let signatureId = template?.signature?.id; + if (!signatureId && application.templateMismatchMode === 'direct_send') { + const signatureName = data.content.match(/^【[^】]+】/)?.[0]; + signatureId = (signaturesByApplication.get(application.id) ?? []).find((signature) => signature.name === signatureName)?.id; + } + if (!signatureId || (!template && application.templateMismatchMode !== 'direct_send')) continue; + const finalSignatureId = signatureId; + candidates.push({ item, payload, application, phoneNumber, template, signatureId: finalSignatureId, templateVariables, clientSrcId }); + } catch { + // Invalid source IDs and other business rejections stay on the established + // individual path so their exact failure receipt remains unchanged. + } + } + if (candidates.length < 2) return new Set(); + + const quota = await this.measureInboundStage('daily_quota', () => this.reserveDailyQuotaBatch(candidates)); + const quotaApproved = candidates.filter((candidate) => quota.get(candidate.item.requestKey)?.reserved); + if (quotaApproved.length < 2) return new Set(); + const riskResults = await this.measureInboundStage('risk_frequency', () => this.riskReview.evaluateTasksBatch(quotaApproved.map((candidate) => ({ + tenantId: candidate.application.tenantId, + applicationId: candidate.application.id, + templateId: candidate.template?.id, + content: candidate.payload.data.content, + variables: candidate.templateVariables, + phones: [candidate.phoneNumber], + sourceType: 'cmpp' as const, + })))); + const riskApproved = quotaApproved.filter((_, index) => riskResults[index]?.status === 'approved'); + if (riskApproved.length < 2) return new Set(); + const frequency = await this.measureInboundStage('risk_frequency', () => this.phoneFrequency.reserveBatch(riskApproved.map((candidate) => ({ + tenantId: candidate.application.tenantId, + applicationId: candidate.application.id, + phoneNumber: candidate.phoneNumber, + sourceType: 'cmpp', + reservationKey: `${candidate.item.requestKey}:message:0:frequency`, + })))); + const approved = riskApproved.filter((candidate) => ( + (frequency.get(`${candidate.item.requestKey}:message:0:frequency`)?.size ?? 0) === 0 + )); + if (approved.length < 2) return new Set(); + + const drainageRows = await this.prisma.smsDrainageInfo.findMany({ + where: { signatureId: { in: [...new Set(approved.map((candidate) => candidate.signatureId))] }, auditStatus: { not: 'deleted' } }, + select: { id: true, signatureId: true, url: true, updatedAt: true }, + orderBy: [{ updatedAt: 'desc' }, { id: 'asc' }], + }); + const finalCandidates = approved.filter((candidate) => { + const matches = drainageRows + .filter((row) => row.signatureId === candidate.signatureId && row.url.trim() && candidate.payload.data.content.includes(row.url.trim())) + .sort((left, right) => right.url.trim().length - left.url.trim().length || right.updatedAt.getTime() - left.updatedAt.getTime()); + if (matches.length > 1 && matches[0].url.trim().length === matches[1].url.trim().length) return false; + candidate.drainageInfoId = matches[0]?.id; + return true; + }); + if (finalCandidates.length < 2) return new Set(); + return this.persistCommonInboundWorkflowBatch(finalCandidates); + } + + private async reserveDailyQuotaBatch(candidates: InboundBatchCandidate[]) { + const usageDate = shanghaiDateKey(); + const usageDateValue = new Date(`${usageDate}T00:00:00.000Z`); + return this.prisma.$transaction(async (tx) => { + const applicationIds = [...new Set(candidates.map((candidate) => candidate.application.id))].sort(); + await tx.smsApplicationDailyUsage.createMany({ + data: applicationIds.map((applicationId) => ({ id: randomUUID(), applicationId, usageDate: usageDateValue, usedCount: 0 })), + skipDuplicates: true, + }); + const usage = await tx.$queryRaw>(Prisma.sql` + SELECT usage."applicationId", application."tenantId", + COALESCE(application."dailyLimit", 100000)::integer AS "dailyLimit", + usage."usedCount" + FROM "SmsApplicationDailyUsage" usage + JOIN "SmsApplication" application ON application.id = usage."applicationId" + WHERE usage."usageDate" = ${usageDate}::date + AND usage."applicationId" IN (${Prisma.join(applicationIds)}) + ORDER BY usage."applicationId" + FOR UPDATE OF usage + `); + const state = new Map(usage.map((row) => [row.applicationId, { ...row }])); + const keys = candidates.map((candidate) => `${candidate.item.requestKey}:daily-quota`); + const existing = await tx.smsApplicationDailyReservation.findMany({ where: { reservationKey: { in: keys } } }); + const existingByKey = new Map(existing.map((row) => [row.reservationKey, row])); + const output = new Map(); + const inserts: Prisma.SmsApplicationDailyReservationCreateManyInput[] = []; + for (const candidate of candidates) { + const reservationKey = `${candidate.item.requestKey}:daily-quota`; + const replay = existingByKey.get(reservationKey); + if (replay) { + if (replay.applicationId !== candidate.application.id || replay.requestedCount !== 1) { + throw new Error('CMPP daily quota idempotency key conflicts with another reservation'); + } + output.set(candidate.item.requestKey, { reserved: replay.reserved, dailyLimit: replay.dailyLimit, usedCount: replay.usedCount }); + continue; + } + const current = state.get(candidate.application.id); + if (!current) throw new Error(`CMPP daily quota application ${candidate.application.id} disappeared`); + const reserved = current.usedCount + 1 <= current.dailyLimit; + if (reserved) current.usedCount += 1; + inserts.push({ + id: randomUUID(), reservationKey, tenantId: candidate.application.tenantId, + applicationId: candidate.application.id, usageDate: usageDateValue, requestedCount: 1, + dailyLimit: current.dailyLimit, usedCount: reserved ? current.usedCount : null, reserved, + }); + output.set(candidate.item.requestKey, { reserved, dailyLimit: current.dailyLimit, usedCount: reserved ? current.usedCount : null }); + } + for (const row of state.values()) { + await tx.smsApplicationDailyUsage.update({ + where: { applicationId_usageDate: { applicationId: row.applicationId, usageDate: usageDateValue } }, + data: { usedCount: row.usedCount }, + }); + } + if (inserts.length) await tx.smsApplicationDailyReservation.createMany({ data: inserts }); + return output; + }, { isolationLevel: Prisma.TransactionIsolationLevel.ReadCommitted }); + } + + private async persistCommonInboundWorkflowBatch(candidates: InboundBatchCandidate[]) { + const prepared = await Promise.all(candidates.map(async (candidate) => { + const workflowDigest = createHash('sha256').update(`${candidate.item.requestKey}:message:0`).digest('hex').slice(0, 32); + const taskId = randomUUID(); + const messageRecordId = randomUUID(); + const content = candidate.payload.data.content; + const drainageDetection = await detectDrainageContent(this.prisma, content); + return { candidate, workflowDigest, taskId, messageRecordId, content, drainageDetection }; + })); + await this.measureInboundStage('message_persist', () => this.prisma.$transaction(async (tx) => { + await tx.smsBatchTask.createMany({ + data: prepared.map(({ candidate, workflowDigest, taskId, content }) => ({ + id: taskId, + tenantId: candidate.application.tenantId, + applicationId: candidate.application.id, + templateId: candidate.template?.id, + taskNo: `BT-IN-${workflowDigest}`, + sourceType: 'cmpp', + content, + phoneTotal: 1, + status: 'ready', + auditStatus: 'approved', + progressTotal: 1, + })), + }); + await tx.smsApiRequest.createMany({ + data: prepared.map(({ candidate, workflowDigest, taskId, content }) => ({ + id: randomUUID(), tenantId: candidate.application.tenantId, batchTaskId: taskId, + requestId: `REQ-IN-${workflowDigest}`, sourceIp: candidate.payload.data.remoteIp, + userAgent: 'cmpp-gateway', payloadSummary: { phoneTotal: 1, contentLength: [...content].length, account: candidate.payload.data.account }, + status: 'accepted', + })), + }); + await tx.smsMessageRecord.createMany({ + data: prepared.map(({ candidate, taskId, messageRecordId, content, drainageDetection }) => ({ + id: messageRecordId, + tenantId: candidate.application.tenantId, + batchTaskId: taskId, + applicationId: candidate.application.id, + templateId: candidate.template?.id, + signatureId: candidate.signatureId, + drainageInfoId: candidate.drainageInfoId, + messageId: candidate.payload.messageIds[0], + phoneNumber: candidate.phoneNumber, + content, + ...drainageDetection, + billingUnits: this.billing.estimateSmsCost({ + tenantId: candidate.application.tenantId, applicationId: candidate.application.id, + content, phoneCount: 1, unitPrice: 0, + }).billingUnitsPerMessage, + unitPrice: 0, + amountCents: 0, + queuePriority: normalizeQueuePriority(candidate.application.queuePriority), + cmppSubmitSequenceId: candidate.payload.data.sequenceId == null ? null : String(candidate.payload.data.sequenceId), + cmppSubmitGroupMessageId: candidate.payload.submitGroupMessageId, + cmppRegisteredDelivery: candidate.payload.data.registeredDelivery !== 0, + clientSrcId: candidate.clientSrcId, + applicationExtension: candidate.application.cmppApplicationExtension, + status: 'queued', + })), + }); + })); + await this.measureInboundStage('queue_publish', () => this.facade.getSendQueue().addBulk(prepared.map(({ candidate, messageRecordId }) => ({ + name: 'send-message' as const, + data: { messageRecordId }, + opts: { + jobId: messageRecordId, + attempts: 3, + priority: BULLMQ_PRIORITY[normalizeQueuePriority(candidate.application.queuePriority)], + }, + })))); + await this.prisma.smsBatchTask.updateMany({ + where: { id: { in: prepared.map((entry) => entry.taskId) }, status: 'ready' }, + data: { status: 'queued' }, + }); + const results = prepared.map(({ candidate, taskId, messageRecordId }) => ({ + accepted: true, + tenantId: candidate.application.tenantId, + applicationId: candidate.application.id, + taskId, + messageId: candidate.payload.messageIds[0], + messageRecordId, + status: 'accepted', + phoneCount: 1, + messages: [{ + phoneNumber: candidate.phoneNumber, + messageId: candidate.payload.messageIds[0], + messageRecordId, + taskId, + status: 'accepted', + }], + })); + const values = prepared.map((entry, index) => Prisma.sql`(${entry.candidate.item.id}, ${JSON.stringify(results[index])}::jsonb)`); + const settled = await this.prisma.$queryRaw>(Prisma.sql` + UPDATE "CmppInboundSubmissionInbox" inbox + SET status = 'completed', result = updates.result, "completedAt" = (NOW() AT TIME ZONE 'UTC'), + "lockedAt" = NULL, "lockedBy" = NULL, "lastError" = NULL, "updatedAt" = (NOW() AT TIME ZONE 'UTC') + FROM (VALUES ${Prisma.join(values)}) AS updates(id, result) + WHERE inbox.id = updates.id + AND inbox.status = 'processing' + AND inbox."lockedBy" = ${this.inboundWorkflowWorkerId} + RETURNING inbox.id + `); + const settledIds = new Set(settled.map((row) => row.id)); + for (let index = 0; index < settled.length; index += 1) this.metrics?.recordInboundWorkflowResult('completed'); + return settledIds; + } + private async processClaimedInboundWorkflow(item: ClaimedInboundWorkflow, application?: InboundApplication) { try { const payload = parseInboundWorkflowPayload(item.payload); diff --git a/docs/codebase-modularization-roadmap.md b/docs/codebase-modularization-roadmap.md index 0470816..3c08e77 100644 --- a/docs/codebase-modularization-roadmap.md +++ b/docs/codebase-modularization-roadmap.md @@ -1161,4 +1161,6 @@ global，不能错误归入client。`client-signature-*`、发送页、企业认 - `api/src/infrastructure-monitoring/`是运营端监控聚合与固定阈值应用边界：只消费代码白名单 PromQL，并通过版本化 PostgreSQL 单例、promtool 校验和原子规则热加载管理数值阈值；Exporter安装、端口隔离、固定规则模板和权限仍归`tools/monitoring/`治理。 - 活动告警已读也归该边界：Prometheus保留告警事实，Prisma仅持久化逐管理员、逐触发周期的阅读状态；全局布局只消费轻量未读汇总，不复制指纹、activeAt或用户隔离逻辑。 - 500条/秒第二阶段仍限制在`send-inbound-entry`和既有进程容量边界：短短信用一个PostgreSQL CTE完成当前应用校验和Inbox写入；长短信继续走既有分片域。Worker可按一次领取批次预取应用快照，但不得引入跨批应用状态缓存，也不得把模板、风控、计费、路由状态塞回Gateway同步入口。 +- 500条/秒第三阶段继续留在`send-inbound-entry`编排边界，但将批次只读风控快照下沉到`RiskReviewService.evaluateTasksBatch`，将频控原子批量预留下沉到`PhoneFrequencyService.reserveBatch`。Inbox编排层只负责批次分组、日限额锁顺序、正常三表批量持久化、幂等入队与逐条租约结算；付费及异常状态机仍委托原单条路径，避免形成第二套计费/审核领域模型。 +- 批次锁顺序固定为：领取事务只锁Inbox后立即提交；日限额事务按applicationId排序锁`SmsApplicationDailyUsage`；频控按tenant/application分组且规则优先级顺序更新号码状态；三表事务不持有前两类锁，Redis/BullMQ发布永不位于数据库事务内。该顺序用于限制死锁面并允许单批失败后按每条稳定幂等键恢复。 - 容量参数继续归进程组装层：Inbox业务槽、BullMQ发送槽、Gateway供应商槽和结果Outbox槽分别有界、分别观测。代码模块不得假定测试环境参数就是生产默认值；生产调优必须重新基于PostgreSQL连接预算、六通道TPS/窗口和回调承载证据。 diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index d781c2f..346edfc 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -2122,3 +2122,10 @@ - 长短信仍在完整重组并完成既有校验后写Inbox，不为追求吞吐改写分片状态机。Inbox Worker每次领取后应按本批应用ID一次读取当前应用/企业/白名单快照，不能对同一批每条消息重复查应用；领取、实际业务处理和逐条完成仍保持原短事务与租约边界。 - 测试环境可在PostgreSQL连接总预算内分别提高Worker业务槽、BullMQ发送槽、Gateway供应商槽及结果Outbox槽，但每个池必须显式有界，且总数据库连接须为运维保留余量。供应商真实TPS、连接数和窗口仍是硬上限，禁止用并发参数绕过单通道限速。 - 第二阶段仍以“完整处理”验收：除500条/秒SubmitResp零拒绝、零缺失和延迟停止线外，还必须等待Inbox、BullMQ、命令Stream、结果Outbox与回执链排空，并按数据库业务记录、唯一MessageId、供应商尝试/接受和最终状态对账。入口达到500而全链排空速率不足500条/秒时必须明确判失败。 + +## 完整处理500条/秒第三阶段：Inbox业务批处理与数据库竞争治理（2026-08-21） + +- Worker以`API_INBOUND_WORKFLOW_BATCH_SIZE`控制单次有界批次，仍用优先级/FIFO和`FOR UPDATE SKIP LOCKED`短事务领取；批次只共享当次读取的应用、模板、签名、黑名单、风控规则和敏感词快照，不得跨批缓存应用状态、余额、频控计数或日限额。 +- 正常单号码、零计费短短信允许走批量快路径：日限额按应用锁定并为每个Inbox请求写独立幂等预留，风控只共享只读输入，号码频控在同一事务批量更新状态并逐请求保存决定，任务/API请求/消息三表在一个短事务批量创建，BullMQ使用消息ID作为幂等Job ID批量入队。 +- 付费短信、重复号码、黑名单/格式拒绝、模板人工审核、引流匹配歧义、既有消息恢复及其他异常分支必须回到原逐条状态机；不得为压测降低计费、风控、频控、模板、签名或失败回执规则。批量路径在数据库提交后、队列发布前崩溃时，逐条恢复必须利用稳定任务号、请求号、MessageId和预留键补齐，不得重复计量或创建消息。 +- 批量Worker指标增加`worker_claim/reference_preload/daily_quota`固定低基数阶段，并继续记录`risk_frequency/message_persist/queue_publish`；配置必须显式启用批处理并使用正整数批次大小。验收仍以真实PostgreSQL、Redis、BullMQ和隔离供应商证据为准，不能用零计费压测结果外推付费链路吞吐。 diff --git a/docs/production-deployment.md b/docs/production-deployment.md index 7cc2747..c827a4f 100644 --- a/docs/production-deployment.md +++ b/docs/production-deployment.md @@ -154,7 +154,7 @@ curl http://127.0.0.1:12026/ redis-cli -h 127.0.0.1 -p 6379 ping pg_isready -d "$(grep '^DATABASE_URL=' /etc/cmpp-platform/cmpp-platform.env | cut -d= -f2-)" grep -E '^(API_ENABLE_SEND_WORKER|API_SEND_WORKER_CONCURRENCY|GATEWAY_SUBMIT_WORKER_CONCURRENCY|GATEWAY_SUBMIT_RESULT_WORKER_CONCURRENCY|GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY)=' /etc/cmpp-platform/cmpp-platform.env -grep -E '^(CMPP_INBOUND_FAST_PATH_ENABLED|CMPP_INBOUND_WORKFLOW_WORKER_ENABLED|API_INBOUND_WORKFLOW_CONCURRENCY|API_INBOUND_WORKFLOW_POLL_INTERVAL_MS|API_INBOUND_WORKFLOW_STALE_SECONDS|API_DB_POOL_MAX|API_WORKER_DB_POOL_MAX|API_WORKER_METRICS_PORT)=' /etc/cmpp-platform/cmpp-platform.env +grep -E '^(CMPP_INBOUND_FAST_PATH_ENABLED|CMPP_INBOUND_WORKFLOW_WORKER_ENABLED|API_INBOUND_WORKFLOW_CONCURRENCY|API_INBOUND_WORKFLOW_BATCH_ENABLED|API_INBOUND_WORKFLOW_BATCH_SIZE|API_INBOUND_WORKFLOW_POLL_INTERVAL_MS|API_INBOUND_WORKFLOW_STALE_SECONDS|API_DB_POOL_MAX|API_WORKER_DB_POOL_MAX|API_WORKER_METRICS_PORT)=' /etc/cmpp-platform/cmpp-platform.env systemctl is-active cmpp-api cmpp-send-worker cmpp-gateway curl -fsS http://127.0.0.1:9465/metrics | grep '^cmpp_worker_inbound_workflow_' redis-cli --scan --pattern 'rate:gateway:channel:*' @@ -186,3 +186,4 @@ bash tools/deploy/production-deploy.sh - 环境必须显式启用`CMPP_INBOUND_FAST_PATH_ENABLED=true`、`CMPP_INBOUND_WORKFLOW_WORKER_ENABLED=true`，并给出正整数`API_INBOUND_WORKFLOW_CONCURRENCY`、`API_DB_POOL_MAX`和`API_WORKER_DB_POOL_MAX`；推荐初始值分别为32槽、API池32、Worker池8、轮询100ms、租约300秒。API systemd角色必须是`api`，Worker角色必须是`worker`；Worker可通过`API_WORKER_DATABASE_URL`使用独立地址，未配置时仍使用同一数据库地址但保持独立进程和有界连接池。发布前必须核对两池之和、其他服务连接与运维余量不超过PostgreSQL`max_connections`。 - Worker日志目录归`cmpp-api:cmpp-security`且仅服务可写，9465只监听回环并加入Prometheus `cmpp-send-worker` target。发布后必须验证两进程均为非root、API/Gateway health、Worker metrics、PostgreSQL/Redis，以及Inbox pending/processing/最老等待可观测。 - 回滚前先停止Gateway、API和Worker，保留故障现场Inbox及日志；如恢复旧数据库备份，必须同时恢复对应源码和环境/systemd资产。不得在回滚时删除pending Inbox或重投真实短信。 +- 第三阶段要求环境显式设置`API_INBOUND_WORKFLOW_BATCH_ENABLED=true`和正整数`API_INBOUND_WORKFLOW_BATCH_SIZE`，初始建议64且不得大于Worker业务槽的可解释倍数。批次增大前必须核对PostgreSQL参数数量、单事务持续时间、Worker RSS和租约时长；付费短信仍走逐条账务锁，不能用零计费批量结果替代付费链路验收。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index be05d8e..fcd276f 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -4701,6 +4701,18 @@ npm run verify:phase8 | TC-GLOBAL-ALERT-002 | 预警菜单跳转 | 分别点击铃铛中的两个菜单项 | 签名项跳转`/admin/signature-retirement`，安全项跳转`/admin/security-detection`，弹层关闭且对应页面读取真实后端数据 | | TC-GLOBAL-ALERT-003 | 域间故障隔离与轻量轮询 | 分别让一个汇总接口失败并观察30秒轮询请求 | 失败域显示0且另一域数据保留；安全预警使用专用汇总接口，不调用完整overview、规则、代理状态或告警大列表 | | TC-DEPLOY-NET-001 | API回环监听边界 | 使用标准生产环境启动API，执行`ss -lnt`并从LAN/Tailscale探测3000端口，同时经Nginx业务入口请求健康接口 | API仅监听`127.0.0.1:3000`，外部不能直连3000；Nginx入口仍正常返回真实API健康结果；部署静态门禁校验`API_HOST`默认值与启动参数一致 | + +## CMPP第三阶段业务批处理专项（2026-08-21） + +| 用例ID | 场景 | 步骤 | 预期 | +| --- | --- | --- | --- | +| TC-CMPP-PERF-P3-001 | 有界批次领取 | 制造priority/normal混合Inbox并并发启动两个Worker | 每次领取不超过配置批次/槽位，使用SKIP LOCKED，无重复领取；priority先于normal且类内FIFO | +| TC-CMPP-PERF-P3-002 | 批次只读预加载 | 同批放入多应用、多内容正常短短信并统计SQL | 应用、模板、签名、黑名单、风控规则和敏感词按批读取，不逐短信重复；下一批重新读取应用状态 | +| TC-CMPP-PERF-P3-003 | 日限额批量原子性 | 在剩余额度边界并发提交并重放相同请求键 | 使用量不超过上限；每条预留决定独立持久化，重放不重复递增，拒绝项走既有失败回执 | +| TC-CMPP-PERF-P3-004 | 号码频控批量原子性 | 唯一号码批量提交、同号重复提交并模拟Worker崩溃重领 | 唯一号码批量更新；同号保留顺序语义并逐条处理；重领返回原决定，无穿透、重复计频或重复命中 | +| TC-CMPP-PERF-P3-005 | 三表与队列崩溃恢复 | 在三表提交后、BullMQ发布前注入失败并重领 | 任务/API请求/消息不重复，稳定MessageId对应唯一消息；以MessageId Job ID补入队后Inbox独立完成 | +| TC-CMPP-PERF-P3-006 | 异常与付费回退 | 覆盖正单价、黑名单、非法号、人工审核、引流歧义和已有消息 | 全部走原逐条状态机，余额冻结和回执语义不变；不得进入零计费批量快路径 | +| TC-CMPP-PERF-P3-007 | 500条/秒阶梯 | 在100.93.204.60依次执行smoke、100/200/300/500并逐级检查数据库、Redis和日志 | 任一级拒绝、缺响应、持续积压、锁等待或对账不一致立即停止；分别报告入口、Inbox完成和完整供应商链速率，不用积压冒充吞吐 | ## Fail2ban 安全检测与人工封禁测试矩阵（2026-08-14） - 本模块必须执行 `docs/fail2ban-assisted-blocking-test-cases-20260814.md` 中 TC-F2B 全量用例，专项用例是本平台功能测试的组成部分，不是可选附录。 diff --git a/docs/testing-progress.md b/docs/testing-progress.md index 65371c9..c0e7568 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -3817,3 +3817,10 @@ git diff --check - 仅在测试环境连接总预算内设置API/Worker池48/32、Inbox业务槽96、BullMQ发送槽96、Gateway Submit槽128、结果Outbox槽32和客户入站上限128；代码默认值未改变，PostgreSQL`max_connections=100`仍保留至少20条非业务池余量。修复后1条/秒低负载9/9受理、P95=41ms，Inbox与短信主记录均9条且MessageId唯一，双Stream归零。 - 正式100条/秒30秒档生成2998条，2998/2998成功，零拒绝、零节流、零连接错误，P50/P95/P99=`34/79/128ms`；相较第一阶段100条/秒P95=147ms，合并SQL在更高Worker竞争下仍降低入口尾延迟。数据库2998条Inbox全部completed、2998条短信主记录及唯一MessageId完全对账，Inbox从首条创建至末条完成约52.1秒，折算约57.5条/秒。 - 该档未通过“完整处理100条/秒”停止线：全部为移动号码并只走`LGST-M-P`主通道，最终产生3863次供应商尝试，其中accepted3118、rejected78、timeout667；回退通道接受846次。命令Stream和结果Outbox从首条接收至最终0/0约136秒，按2998条业务短信折算约22.0条/秒；最终消息delivered2797、failed70、submitted131。高峰期间API后台协议日志、连接状态和待投递查询出现超时，说明提高Worker/Outbox槽后共享API/数据库回调仍受争用。因100档完整链已经失败，按阶梯停止线没有继续200/500档，不能宣称完整500条/秒达标。 + +## 2026-08-21 完整处理500条/秒第三阶段：业务Worker批处理与数据库竞争治理（实施中） + +- 接管复核：本地`HEAD=57192b7586b1e2c14f2edf0633d2d35936352879`、`origin/main=c4f36fc50d7906dfb2f97c881e9ea43c6a64c370`，ahead 15；测试环境标记`f4560479...+workspace.p2fix.51553e0d5555`，预生产标记`433b2ee5...+gateway-v2.cd7bb8d05e7b`。测试机API/Worker/Gateway/PostgreSQL/Redis/Nginx/Prometheus均active，92条migration、数据库6连接/无idle in transaction/无未授予锁，Inbox和双Stream均排空；预生产仅只读核验，未修改。 +- 当前实现把Worker按默认64条组成有界业务批次：批次共享应用、模板、签名、黑名单、风控规则和敏感词只读快照；日限额按applicationId固定顺序锁定并为每条请求写独立预留；唯一号码频控按应用在短事务批量更新并逐条持久化决定；正常零计费单号码的任务、API请求和消息三表在一个短事务批量创建，再以MessageId作为BullMQ Job ID批量入队和逐条结算Inbox。 +- 正单价、同号重复、业务拒绝、人工审核、歧义匹配、既有消息和其他异常保留原逐条状态机。批量路径任一步失败会使用稳定日限/频控键、确定性任务号/请求号、MessageId和队列Job ID逐条恢复；Redis发布不包在数据库事务中。新增固定阶段`worker_claim/reference_preload/daily_quota`并保留`risk_frequency/message_persist/queue_publish`。 +- 当前本地API正式TypeScript编译通过；RiskReview、PhoneFrequency、SendChain专项3套145项通过。尚未提交、建立本次恢复资产、部署或压测；后续结果必须按smoke→100→200→300→500停止线补记，且零计费批量结果不得外推为付费链路吞吐。 diff --git a/tools/deploy/production-bootstrap.sh b/tools/deploy/production-bootstrap.sh index d2ed6fb..33e8cad 100644 --- a/tools/deploy/production-bootstrap.sh +++ b/tools/deploy/production-bootstrap.sh @@ -16,6 +16,8 @@ API_WORKER_METRICS_PORT="${API_WORKER_METRICS_PORT:-9465}" CMPP_INBOUND_FAST_PATH_ENABLED="${CMPP_INBOUND_FAST_PATH_ENABLED:-true}" CMPP_INBOUND_WORKFLOW_WORKER_ENABLED="${CMPP_INBOUND_WORKFLOW_WORKER_ENABLED:-true}" 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_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}" @@ -205,6 +207,8 @@ API_WORKER_METRICS_PORT=${API_WORKER_METRICS_PORT} CMPP_INBOUND_FAST_PATH_ENABLED=${CMPP_INBOUND_FAST_PATH_ENABLED} 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_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 979038e..4ce5455 100644 --- a/tools/deploy/production-deploy.sh +++ b/tools/deploy/production-deploy.sh @@ -38,6 +38,10 @@ if [[ ! "${API_INBOUND_WORKFLOW_CONCURRENCY:-}" =~ ^[1-9][0-9]*$ ]]; then echo "API_INBOUND_WORKFLOW_CONCURRENCY must be a positive integer in $ENV_FILE." >&2 exit 1 fi +if [[ "${API_INBOUND_WORKFLOW_BATCH_ENABLED:-}" != "true" || ! "${API_INBOUND_WORKFLOW_BATCH_SIZE:-}" =~ ^[1-9][0-9]*$ ]]; then + 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_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 3349301..bfce981 100644 --- a/tools/deploy/verify-production-deployment.mjs +++ b/tools/deploy/verify-production-deployment.mjs @@ -31,6 +31,8 @@ for (const marker of [ 'CMPP_INBOUND_FAST_PATH_ENABLED=true', 'CMPP_INBOUND_WORKFLOW_WORKER_ENABLED=true', 'API_INBOUND_WORKFLOW_CONCURRENCY', + 'API_INBOUND_WORKFLOW_BATCH_ENABLED=true', + 'API_INBOUND_WORKFLOW_BATCH_SIZE', 'cmpp-send-worker.service', 'Environment=CMPP_PROCESS_ROLE=api', 'Environment=CMPP_PROCESS_ROLE=worker',