diff --git a/.env.example b/.env.example index e93f6a5..88098ec 100644 --- a/.env.example +++ b/.env.example @@ -12,6 +12,14 @@ HTTP_API_MASTER_KEY=replace-with-at-least-32-random-characters HTTP_API_PUBLIC_ORIGIN=https://api.example.com API_ENABLE_SEND_WORKER=true API_SEND_WORKER_CONCURRENCY=50 +API_WORKER_DATABASE_URL= +API_WORKER_METRICS_HOST=127.0.0.1 +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_POLL_INTERVAL_MS=100 +API_INBOUND_WORKFLOW_STALE_SECONDS=300 ADMIN_SESSION_IDLE_TIMEOUT_MS=3600000 CLIENT_SESSION_IDLE_TIMEOUT_MS=7200000 SESSION_LOCK_RECOVERY_MS=14400000 diff --git a/api/prisma/migrations/20260820170000_add_cmpp_inbound_submission_inbox/migration.sql b/api/prisma/migrations/20260820170000_add_cmpp_inbound_submission_inbox/migration.sql new file mode 100644 index 0000000..e187211 --- /dev/null +++ b/api/prisma/migrations/20260820170000_add_cmpp_inbound_submission_inbox/migration.sql @@ -0,0 +1,91 @@ +CREATE TABLE "CmppInboundSubmissionInbox" ( + "id" TEXT NOT NULL, + "requestKey" TEXT NOT NULL, + "payloadHash" TEXT NOT NULL, + "tenantId" TEXT NOT NULL, + "applicationId" TEXT NOT NULL, + "queuePriority" TEXT NOT NULL DEFAULT 'normal', + "payload" JSONB NOT NULL, + "response" JSONB NOT NULL, + "status" TEXT NOT NULL DEFAULT 'pending', + "attempts" INTEGER NOT NULL DEFAULT 0, + "nextAttemptAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "lockedAt" TIMESTAMP(3), + "lockedBy" TEXT, + "lastError" TEXT, + "result" JSONB, + "completedAt" TIMESTAMP(3), + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "CmppInboundSubmissionInbox_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX "CmppInboundSubmissionInbox_requestKey_key" +ON "CmppInboundSubmissionInbox"("requestKey"); + +CREATE INDEX "CmppInboundSubmissionInbox_status_nextAttemptAt_createdAt_idx" +ON "CmppInboundSubmissionInbox"("status", "nextAttemptAt", "createdAt"); + +CREATE INDEX "CmppInboundSubmissionInbox_status_lockedAt_idx" +ON "CmppInboundSubmissionInbox"("status", "lockedAt"); + +CREATE INDEX "CmppInboundSubmissionInbox_applicationId_createdAt_idx" +ON "CmppInboundSubmissionInbox"("applicationId", "createdAt"); + +CREATE INDEX "CmppInboundSubmissionInbox_queuePriority_status_createdAt_idx" +ON "CmppInboundSubmissionInbox"("queuePriority", "status", "createdAt"); + +ALTER TABLE "CmppInboundSubmissionInbox" +ADD CONSTRAINT "CmppInboundSubmissionInbox_tenantId_fkey" +FOREIGN KEY ("tenantId") REFERENCES "Tenant"("id") ON DELETE RESTRICT ON UPDATE CASCADE; + +ALTER TABLE "CmppInboundSubmissionInbox" +ADD CONSTRAINT "CmppInboundSubmissionInbox_applicationId_fkey" +FOREIGN KEY ("applicationId") REFERENCES "SmsApplication"("id") ON DELETE RESTRICT ON UPDATE CASCADE; + +CREATE TABLE "SmsApplicationDailyReservation" ( + "id" TEXT NOT NULL, + "reservationKey" TEXT NOT NULL, + "tenantId" TEXT NOT NULL, + "applicationId" TEXT NOT NULL, + "usageDate" DATE NOT NULL, + "requestedCount" INTEGER NOT NULL, + "dailyLimit" INTEGER NOT NULL, + "usedCount" INTEGER, + "reserved" BOOLEAN NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT "SmsApplicationDailyReservation_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX "SmsApplicationDailyReservation_reservationKey_key" +ON "SmsApplicationDailyReservation"("reservationKey"); +CREATE INDEX "SmsApplicationDailyReservation_applicationId_usageDate_idx" +ON "SmsApplicationDailyReservation"("applicationId", "usageDate"); +CREATE INDEX "SmsApplicationDailyReservation_tenantId_createdAt_idx" +ON "SmsApplicationDailyReservation"("tenantId", "createdAt"); +ALTER TABLE "SmsApplicationDailyReservation" ADD CONSTRAINT "SmsApplicationDailyReservation_tenantId_fkey" +FOREIGN KEY ("tenantId") REFERENCES "Tenant"("id") ON DELETE RESTRICT ON UPDATE CASCADE; +ALTER TABLE "SmsApplicationDailyReservation" ADD CONSTRAINT "SmsApplicationDailyReservation_applicationId_fkey" +FOREIGN KEY ("applicationId") REFERENCES "SmsApplication"("id") ON DELETE RESTRICT ON UPDATE CASCADE; + +CREATE TABLE "PhoneFrequencyReservation" ( + "id" TEXT NOT NULL, + "reservationKey" TEXT NOT NULL, + "tenantId" TEXT NOT NULL, + "applicationId" TEXT NOT NULL, + "result" JSONB NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT "PhoneFrequencyReservation_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX "PhoneFrequencyReservation_reservationKey_key" +ON "PhoneFrequencyReservation"("reservationKey"); +CREATE INDEX "PhoneFrequencyReservation_applicationId_createdAt_idx" +ON "PhoneFrequencyReservation"("applicationId", "createdAt"); +CREATE INDEX "PhoneFrequencyReservation_tenantId_createdAt_idx" +ON "PhoneFrequencyReservation"("tenantId", "createdAt"); +ALTER TABLE "PhoneFrequencyReservation" ADD CONSTRAINT "PhoneFrequencyReservation_tenantId_fkey" +FOREIGN KEY ("tenantId") REFERENCES "Tenant"("id") ON DELETE RESTRICT ON UPDATE CASCADE; +ALTER TABLE "PhoneFrequencyReservation" ADD CONSTRAINT "PhoneFrequencyReservation_applicationId_fkey" +FOREIGN KEY ("applicationId") REFERENCES "SmsApplication"("id") ON DELETE RESTRICT ON UPDATE CASCADE; diff --git a/api/prisma/schema.prisma b/api/prisma/schema.prisma index ddbd519..13c2288 100644 --- a/api/prisma/schema.prisma +++ b/api/prisma/schema.prisma @@ -52,6 +52,9 @@ model Tenant { openApiRequests OpenApiRequest[] httpWebhookEvents HttpWebhookEvent[] cmppInboundLongMessages CmppInboundLongMessage[] + cmppInboundSubmissionInboxes CmppInboundSubmissionInbox[] + smsApplicationDailyReservations SmsApplicationDailyReservation[] + phoneFrequencyReservations PhoneFrequencyReservation[] } model EnterpriseCertification { @@ -467,10 +470,13 @@ model SmsApplication { httpWebhookEndpoints HttpWebhookEndpoint[] httpWebhookEvents HttpWebhookEvent[] dailyUsages SmsApplicationDailyUsage[] + dailyReservations SmsApplicationDailyReservation[] inboundLongMessages CmppInboundLongMessage[] + inboundSubmissionInboxes CmppInboundSubmissionInbox[] riskRules RiskRule[] phoneFrequencyStates PhoneFrequencyState[] phoneFrequencyHits PhoneFrequencyHit[] + phoneFrequencyReservations PhoneFrequencyReservation[] @@index([tenantId, status]) @@index([status, createdAt]) @@ -533,6 +539,25 @@ model SmsApplicationDailyUsage { @@index([usageDate]) } +model SmsApplicationDailyReservation { + id String @id @default(cuid()) + reservationKey String @unique + tenantId String + applicationId String + usageDate DateTime @db.Date + requestedCount Int + dailyLimit Int + usedCount Int? + reserved Boolean + createdAt DateTime @default(now()) + + tenant Tenant @relation(fields: [tenantId], references: [id]) + application SmsApplication @relation(fields: [applicationId], references: [id], onDelete: Restrict) + + @@index([applicationId, usageDate]) + @@index([tenantId, createdAt]) +} + model SmsApplicationHttpIpAllowlist { id String @id @default(cuid()) applicationId String @@ -1578,6 +1603,21 @@ model PhoneFrequencyHit { @@index([releasedById]) } +model PhoneFrequencyReservation { + id String @id @default(cuid()) + reservationKey String @unique + tenantId String + applicationId String + result Json + createdAt DateTime @default(now()) + + tenant Tenant @relation(fields: [tenantId], references: [id]) + application SmsApplication @relation(fields: [applicationId], references: [id], onDelete: Restrict) + + @@index([applicationId, createdAt]) + @@index([tenantId, createdAt]) +} + model PhoneFrequencyWhitelist { id String @id @default(cuid()) phoneNumber String @unique @@ -1952,6 +1992,35 @@ model CmppInboundLongMessageSegment { @@index([sequenceId]) } +model CmppInboundSubmissionInbox { + id String @id @default(cuid()) + requestKey String @unique + payloadHash String + tenantId String + applicationId String + queuePriority String @default("normal") + payload Json + response Json + status String @default("pending") + attempts Int @default(0) + nextAttemptAt DateTime @default(now()) + lockedAt DateTime? + lockedBy String? + lastError String? + result Json? + completedAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + tenant Tenant @relation(fields: [tenantId], references: [id]) + application SmsApplication @relation(fields: [applicationId], references: [id]) + + @@index([status, nextAttemptAt, createdAt]) + @@index([status, lockedAt]) + @@index([applicationId, createdAt]) + @@index([queuePriority, status, createdAt]) +} + model SmsReceiptRecord { id String @id @default(cuid()) tenantId String? diff --git a/api/src/metrics/metrics.service.ts b/api/src/metrics/metrics.service.ts index 231365a..c67a050 100644 --- a/api/src/metrics/metrics.service.ts +++ b/api/src/metrics/metrics.service.ts @@ -6,6 +6,7 @@ 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' | 'long_message_fragment' | 'submission_precheck' | 'template_match' @@ -45,6 +46,12 @@ export class MetricsService implements OnModuleDestroy { private readonly http = new Map(); private readonly cmppInbound = new Map(); private inFlight = 0; + private inboundWorkflowPending = 0; + private inboundWorkflowProcessing = 0; + private inboundWorkflowOldestPendingAgeSeconds = 0; + private inboundWorkflowConfiguredSlots = 0; + private inboundWorkflowInFlightSlots = 0; + private readonly inboundWorkflowResults = new Map(); constructor() { this.eventLoopDelay.enable(); @@ -91,6 +98,21 @@ export class MetricsService implements OnModuleDestroy { this.cmppInbound.set(key, metric); } + setInboundWorkflowState(pending: number, processing: number, oldestPendingAgeSeconds: number) { + this.inboundWorkflowPending = Math.max(0, pending); + this.inboundWorkflowProcessing = Math.max(0, processing); + this.inboundWorkflowOldestPendingAgeSeconds = Math.max(0, oldestPendingAgeSeconds); + } + + setInboundWorkflowSlots(configured: number, inFlight: number) { + this.inboundWorkflowConfiguredSlots = Math.max(0, configured); + this.inboundWorkflowInFlightSlots = Math.max(0, inFlight); + } + + recordInboundWorkflowResult(result: 'completed' | 'retry') { + this.inboundWorkflowResults.set(result, (this.inboundWorkflowResults.get(result) ?? 0) + 1); + } + render() { const memory = process.memoryUsage(); const uptime = Number(process.hrtime.bigint() - this.startedAt) / 1_000_000_000; @@ -119,6 +141,19 @@ export class MetricsService implements OnModuleDestroy { '# TYPE cmpp_api_http_request_duration_seconds histogram', '# HELP cmpp_api_cmpp_inbound_stage_duration_seconds CMPP inbound processing duration by bounded stage and result.', '# TYPE cmpp_api_cmpp_inbound_stage_duration_seconds histogram', + '# HELP cmpp_worker_inbound_workflow_items Current durable CMPP inbound workflow items by state.', + '# TYPE cmpp_worker_inbound_workflow_items gauge', + metricLine('cmpp_worker_inbound_workflow_items', this.inboundWorkflowPending, { state: 'pending' }), + metricLine('cmpp_worker_inbound_workflow_items', this.inboundWorkflowProcessing, { state: 'processing' }), + '# HELP cmpp_worker_inbound_workflow_slots Durable workflow worker slots by state.', + '# TYPE cmpp_worker_inbound_workflow_slots gauge', + metricLine('cmpp_worker_inbound_workflow_slots', this.inboundWorkflowConfiguredSlots, { state: 'configured' }), + metricLine('cmpp_worker_inbound_workflow_slots', this.inboundWorkflowInFlightSlots, { state: 'in_flight' }), + '# HELP cmpp_worker_inbound_workflow_oldest_pending_age_seconds Age of the oldest pending durable workflow.', + '# TYPE cmpp_worker_inbound_workflow_oldest_pending_age_seconds gauge', + metricLine('cmpp_worker_inbound_workflow_oldest_pending_age_seconds', this.inboundWorkflowOldestPendingAgeSeconds), + '# HELP cmpp_worker_inbound_workflow_results_total Durable workflow processing outcomes.', + '# TYPE cmpp_worker_inbound_workflow_results_total counter', ]; for (const [key, metric] of this.http) { const [method, route, status] = key.split('\u0000'); @@ -141,6 +176,9 @@ export class MetricsService implements OnModuleDestroy { lines.push(metricLine('cmpp_api_cmpp_inbound_stage_duration_seconds_sum', metric.durationSum, labels)); lines.push(metricLine('cmpp_api_cmpp_inbound_stage_duration_seconds_count', metric.count, labels)); } + for (const [result, count] of this.inboundWorkflowResults) { + lines.push(metricLine('cmpp_worker_inbound_workflow_results_total', count, { result })); + } this.eventLoopDelay.reset(); return `${lines.join('\n')}\n`; } diff --git a/api/src/prisma/prisma.service.ts b/api/src/prisma/prisma.service.ts index 8317e5a..c8cad92 100644 --- a/api/src/prisma/prisma.service.ts +++ b/api/src/prisma/prisma.service.ts @@ -6,9 +6,12 @@ import { requestContext } from '../common/request-context'; @Injectable() export class PrismaService extends PrismaClient implements OnModuleDestroy { constructor() { + const databaseUrl = process.env.CMPP_PROCESS_ROLE === 'worker' + ? process.env.API_WORKER_DATABASE_URL || process.env.DATABASE_URL + : process.env.DATABASE_URL; super({ adapter: new PrismaPg( - process.env.DATABASE_URL ?? + databaseUrl ?? 'postgresql://cmpp:cmpp_password@localhost:5432/cmpp_platform?schema=public', ), }); diff --git a/api/src/risk-review/phone-frequency.service.ts b/api/src/risk-review/phone-frequency.service.ts index 608b98d..06fb866 100644 --- a/api/src/risk-review/phone-frequency.service.ts +++ b/api/src/risk-review/phone-frequency.service.ts @@ -80,6 +80,7 @@ export class PhoneFrequencyService { phones: string[], sourceType?: string, requestedAt = new Date(), + reservationKey?: string, ) { if (!applicationId) return new Map(); const normalizedPhones = [...new Set(phones.map((phone) => phone.trim()).filter(Boolean))].sort(); @@ -87,14 +88,35 @@ export class PhoneFrequencyService { await this.riskReview.ensureDefaultRules(); const rules = await this.effectiveRules(applicationId); - if (rules.length === 0) return new Map(); + const normalizedReservationKey = reservationKey?.trim(); return this.prisma.$transaction(async (tx) => { + if (normalizedReservationKey) { + // Frequency counters and the reservation result commit together. Retrying a reclaimed + // Inbox item therefore returns the original decision without incrementing either window. + await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${'phone-frequency:' + normalizedReservationKey}, 0))`; + const existingReservation = await tx.phoneFrequencyReservation.findUnique({ + where: { reservationKey: normalizedReservationKey }, + }); + if (existingReservation) { + if (existingReservation.tenantId !== tenantId || existingReservation.applicationId !== applicationId) { + throw new BadRequestException('号码频控幂等键已用于另一笔预留'); + } + return frequencyRejectionsFromJson(existingReservation.result); + } + } const rejected = new Map(); // 平台级白名单只截断号码频控链路；调用 reserve 之前已执行的格式、黑名单等校验不受影响。 const whitelistedPhones = await this.findActiveWhitelistedPhones(tx, normalizedPhones); const controlledPhones = normalizedPhones.filter((phone) => !whitelistedPhones.has(phone)); - if (controlledPhones.length === 0) return rejected; + if (controlledPhones.length === 0 || rules.length === 0) { + if (normalizedReservationKey) { + await tx.phoneFrequencyReservation.create({ + data: { reservationKey: normalizedReservationKey, tenantId, applicationId, result: [] }, + }); + } + return rejected; + } for (const rule of rules) { const window = fixedShanghaiWindow(requestedAt, readPeriodSeconds(rule)); // 分块限制 SQL 参数数量，但两条规则的全部分块仍在同一事务中提交或回滚。 @@ -144,6 +166,16 @@ export class PhoneFrequencyService { } } } + if (normalizedReservationKey) { + await tx.phoneFrequencyReservation.create({ + data: { + reservationKey: normalizedReservationKey, + tenantId, + applicationId, + result: [...rejected.entries()].map(([phoneNumber, rejection]) => ({ phoneNumber, ...rejection })), + }, + }); + } return rejected; }, { isolationLevel: Prisma.TransactionIsolationLevel.ReadCommitted }); } @@ -566,6 +598,18 @@ export class PhoneFrequencyService { } } +function frequencyRejectionsFromJson(value: Prisma.JsonValue) { + const result = new Map(); + if (!Array.isArray(value)) return result; + for (const item of value) { + if (!item || typeof item !== 'object' || Array.isArray(item)) continue; + const phoneNumber = typeof item.phoneNumber === 'string' ? item.phoneNumber : ''; + const reason = typeof item.reason === 'string' ? item.reason : ''; + if (phoneNumber && reason) result.set(phoneNumber, { code: 'PHONE_FREQUENCY_LIMIT', reason }); + } + return result; +} + const whitelistUserInclude = { createdBy: { select: { id: true, username: true, displayName: true } }, updatedBy: { select: { id: true, username: true, displayName: true } }, diff --git a/api/src/send-chain/send-batch-entry.service.ts b/api/src/send-chain/send-batch-entry.service.ts index 2231cba..c984fc6 100644 --- a/api/src/send-chain/send-batch-entry.service.ts +++ b/api/src/send-chain/send-batch-entry.service.ts @@ -1,4 +1,4 @@ -import { BadRequestException, HttpException, HttpStatus, Logger, NotFoundException } from '@nestjs/common'; +import { BadRequestException, ConflictException, HttpException, HttpStatus, Logger, NotFoundException } from '@nestjs/common'; import { Prisma } from '@prisma/client'; import { Queue, Worker } from 'bullmq'; import IORedis from 'ioredis'; @@ -525,15 +525,16 @@ async reserveDailySendQuota(applicationId: string, requestedCount: number) { return result; } -async tryReserveDailySendQuota(applicationId: string, requestedCount: number) { +async tryReserveDailySendQuota(applicationId: string, requestedCount: number, reservationKey?: string) { if (!Number.isInteger(requestedCount) || requestedCount <= 0) { throw new BadRequestException('发送号码数量必须为正整数'); } const usageDate = shanghaiDateKey(); - const reservationId = randomUUID(); - const rows = await this.prisma.$queryRaw>(Prisma.sql` + const reserve = (client: Pick) => { + const reservationId = randomUUID(); + return client.$queryRaw>(Prisma.sql` WITH application_limit AS ( - SELECT id, COALESCE("dailyLimit", 100000)::integer AS "dailyLimit" + SELECT id, "tenantId", COALESCE("dailyLimit", 100000)::integer AS "dailyLimit" FROM "SmsApplication" WHERE id = ${applicationId} ), reservation AS ( @@ -550,10 +551,49 @@ async tryReserveDailySendQuota(applicationId: string, requestedCount: number) { <= (SELECT "dailyLimit" FROM application_limit) RETURNING "usedCount" ) - SELECT application_limit."dailyLimit", reservation."usedCount" + SELECT application_limit."tenantId", application_limit."dailyLimit", reservation."usedCount" FROM application_limit LEFT JOIN reservation ON TRUE - `); + `); + }; + const normalizedReservationKey = reservationKey?.trim(); + const rows = normalizedReservationKey + ? await this.prisma.$transaction(async (tx) => { + // The quota increment and its idempotency record share one short transaction. A worker + // crash can therefore neither lose a successful reservation nor increment it twice. + await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${'daily-quota:' + normalizedReservationKey}, 0))`; + const existing = await tx.smsApplicationDailyReservation.findUnique({ + where: { reservationKey: normalizedReservationKey }, + }); + if (existing) { + if (existing.applicationId !== applicationId || existing.requestedCount !== requestedCount) { + throw new ConflictException('日发送配额幂等键已用于另一笔预留'); + } + return [{ + tenantId: existing.tenantId, + dailyLimit: existing.dailyLimit, + usedCount: existing.usedCount, + }]; + } + const reservedRows = await reserve(tx); + if (reservedRows.length > 0) { + const row = reservedRows[0]; + await tx.smsApplicationDailyReservation.create({ + data: { + reservationKey: normalizedReservationKey, + tenantId: row.tenantId, + applicationId, + usageDate, + requestedCount, + dailyLimit: Number(row.dailyLimit), + usedCount: row.usedCount == null ? null : Number(row.usedCount), + reserved: row.usedCount != null, + }, + }); + } + return reservedRows; + }) + : await reserve(this.prisma); if (rows.length === 0) { throw new NotFoundException('短信应用不存在'); } diff --git a/api/src/send-chain/send-chain.contracts.ts b/api/src/send-chain/send-chain.contracts.ts index 4ef2b0d..d77c85a 100644 --- a/api/src/send-chain/send-chain.contracts.ts +++ b/api/src/send-chain/send-chain.contracts.ts @@ -30,6 +30,7 @@ export interface GatewayInboundAuthDto { } export interface GatewayInboundSubmitDto { + requestId?: string; account: string; phoneNumber?: string; phoneNumbers?: string[]; diff --git a/api/src/send-chain/send-chain.service.spec.ts b/api/src/send-chain/send-chain.service.spec.ts index ce7756a..ec3342c 100644 --- a/api/src/send-chain/send-chain.service.spec.ts +++ b/api/src/send-chain/send-chain.service.spec.ts @@ -329,6 +329,13 @@ function createPrismaMock() { lastError: 'downstream client is not connected', }), }, + cmppInboundSubmissionInbox: { + create: jest.fn().mockResolvedValue({ id: 'inbox-1' }), + findUnique: jest.fn().mockResolvedValue(null), + updateMany: jest.fn().mockResolvedValue({ count: 1 }), + count: jest.fn().mockResolvedValue(0), + findFirst: jest.fn().mockResolvedValue(null), + }, smsBillingRecord: { findFirst: jest.fn().mockResolvedValue(null), create: jest.fn().mockResolvedValue({ id: 'bill-1' }), @@ -2098,6 +2105,95 @@ describe('SendChainService', () => { }); }); + it('returns from the durable CMPP Inbox fast path before risk, billing, or queue publication', async () => { + const previous = process.env.CMPP_INBOUND_FAST_PATH_ENABLED; + process.env.CMPP_INBOUND_FAST_PATH_ENABLED = 'true'; + try { + const { service, prisma, billing, riskReview, phoneFrequency } = createService(); + service.enqueueBatchTask = jest.fn(); + + const result = await service.submitInboundMessage({ + requestId: 'cmpp-inbound:test-fast-path', + account: '100001', + phoneNumbers: ['13800000001', '13900000002'], + content: 'hello', + sequenceId: 777, + remoteIp: '127.0.0.1', + }); + + expect(result).toEqual(expect.objectContaining({ + accepted: true, + status: 'accepted_pending', + phoneCount: 2, + })); + expect(prisma.cmppInboundSubmissionInbox.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + requestKey: 'cmpp-inbound:test-fast-path', + tenantId: 'tenant-1', + applicationId: 'app-1', + }), + }); + expect(prisma.smsBatchTask.create).not.toHaveBeenCalled(); + expect(prisma.smsMessageRecord.create).not.toHaveBeenCalled(); + expect(riskReview.evaluateTask).not.toHaveBeenCalled(); + expect(phoneFrequency.reserve).not.toHaveBeenCalled(); + expect(billing.freeze).not.toHaveBeenCalled(); + expect(service.enqueueBatchTask).not.toHaveBeenCalled(); + } finally { + if (previous == null) delete process.env.CMPP_INBOUND_FAST_PATH_ENABLED; + else process.env.CMPP_INBOUND_FAST_PATH_ENABLED = previous; + } + }); + + it('returns the stored SubmitResp for an idempotent CMPP Inbox retry', async () => { + const previous = process.env.CMPP_INBOUND_FAST_PATH_ENABLED; + process.env.CMPP_INBOUND_FAST_PATH_ENABLED = 'true'; + try { + const { service, prisma } = createService(); + let originalHash = ''; + const storedResponse = { + accepted: true, + tenantId: 'tenant-1', + applicationId: 'app-1', + taskId: '', + messageId: 'MSG-stable', + messageRecordId: '', + status: 'accepted_pending', + phoneCount: 1, + messages: [{ phoneNumber: '13800000001', messageId: 'MSG-stable', messageRecordId: '', taskId: '', status: 'accepted_pending' }], + }; + prisma.cmppInboundSubmissionInbox.create + .mockImplementationOnce(({ data }) => { + originalHash = data.payloadHash; + return Promise.resolve({ id: 'inbox-1' }); + }) + .mockRejectedValueOnce(new Prisma.PrismaClientKnownRequestError('duplicate request key', { + code: 'P2002', + clientVersion: '7.9.0', + })); + prisma.cmppInboundSubmissionInbox.findUnique.mockImplementation(() => Promise.resolve({ + payloadHash: originalHash, + response: storedResponse, + })); + const request = { + requestId: 'cmpp-inbound:test-retry', + account: '100001', + phoneNumber: '13800000001', + content: 'hello', + sequenceId: 778, + remoteIp: '127.0.0.1', + }; + + await service.submitInboundMessage(request); + await expect(service.submitInboundMessage(request)).resolves.toEqual(storedResponse); + expect(prisma.cmppInboundSubmissionInbox.create).toHaveBeenCalledTimes(2); + expect(prisma.smsMessageRecord.create).not.toHaveBeenCalled(); + } finally { + if (previous == null) delete process.env.CMPP_INBOUND_FAST_PATH_ENABLED; + else process.env.CMPP_INBOUND_FAST_PATH_ENABLED = previous; + } + }); + it('enqueues a freshly persisted inbound message without querying the task and message again', async () => { const { service, prisma } = createService(); const add = jest.fn().mockResolvedValue(undefined); diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index ddf3000..b52a2b5 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -78,9 +78,14 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { } onModuleInit() { + const processRole = process.env.CMPP_PROCESS_ROLE?.trim() || 'all'; + if (processRole === 'api') return; if (process.env.API_ENABLE_SEND_WORKER === 'true') { this.startWorker(); } + if (process.env.CMPP_INBOUND_WORKFLOW_WORKER_ENABLED === 'true') { + this.submission.startInboundWorkflowWorker(); + } if (process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED !== 'false') { this.receiptTimeoutInitialTimer = setTimeout(() => void this.runReceiptTimeoutScan(), RECEIPT_TIMEOUT_INITIAL_DELAY_MS); this.receiptTimeoutInitialTimer.unref?.(); @@ -159,6 +164,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { await this.sendQueue?.close(); await this.gatewayQueue?.close(); this.redis?.disconnect(); + await this.submission.onModuleDestroy(); } async createBatchTask(data: CreateBatchTaskDto) { @@ -579,8 +585,10 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { phoneNumbers: string[], application: Awaited>, requestedGroupMessageId?: string, + requestedMessageIds?: string[], + workflowKey?: string, ) { - return this.submission.submitCompleteInboundMessage(data, phoneNumbers, application, requestedGroupMessageId); + return this.submission.submitCompleteInboundMessage(data, phoneNumbers, application, requestedGroupMessageId, requestedMessageIds, workflowKey); } private async collectInboundLongMessageFragment( @@ -602,8 +610,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { application: NonNullable>>, synchronousRejection?: { code: string; reason: string }, receiptRejection?: { code: string; reason: string }, + workflowItemKey?: string, ) { - return this.submission.submitInboundSingleMessage(data, messageId, submitGroupMessageId, application, synchronousRejection, receiptRejection); + return this.submission.submitInboundSingleMessage(data, messageId, submitGroupMessageId, application, synchronousRejection, receiptRejection, workflowItemKey); } /** @@ -618,8 +627,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { variables?: Record; phoneNumber: string; sourceType: 'cmpp'; - }) { - return this.submission.evaluateRiskWithPhoneFrequency(input); + }, reservationKey?: string) { + return this.submission.evaluateRiskWithPhoneFrequency(input, reservationKey); } async markUnknownTimeout(data: TimeoutUnknownDto) { @@ -774,8 +783,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { return this.submission.reserveDailySendQuota(applicationId, requestedCount); } - private async tryReserveDailySendQuota(applicationId: string, requestedCount: number) { - return this.submission.tryReserveDailySendQuota(applicationId, requestedCount); + private async tryReserveDailySendQuota(applicationId: string, requestedCount: number, reservationKey?: string) { + return this.submission.tryReserveDailySendQuota(applicationId, requestedCount, reservationKey); } private async chargeAcceptedMessage(message: { diff --git a/api/src/send-chain/send-inbound-entry.service.ts b/api/src/send-chain/send-inbound-entry.service.ts index a587a81..1080410 100644 --- a/api/src/send-chain/send-inbound-entry.service.ts +++ b/api/src/send-chain/send-inbound-entry.service.ts @@ -3,6 +3,7 @@ import { Prisma } from '@prisma/client'; import { Queue, Worker } from 'bullmq'; import IORedis from 'ioredis'; import { createHash, randomUUID } from 'node:crypto'; +import { hostname } from 'node:os'; import { setTimeout as sleep } from 'node:timers/promises'; import { BillingService } from '../billing/billing.service'; import { isIpAllowed } from '../common/ip-allowlist'; @@ -17,11 +18,33 @@ import { SEND_QUEUE, GATEWAY_SUBMIT_QUEUE, GATEWAY_SUBMIT_STREAM, DEFAULT_SCHEDU import type { SendSubmissionCallbacks, SendSubmissionService } from './send-submission.service'; import { detectDrainageContent } from './drainage-content-detection'; +type InboundWorkflowPayload = { + data: GatewayInboundSubmitDto; + phoneNumbers: string[]; + submitGroupMessageId: string; + messageIds: string[]; +}; + +type ClaimedInboundWorkflow = { + id: string; + requestKey: string; + applicationId: string; + attempts: number; + payload: Prisma.JsonValue; +}; + /** * R9 inboundEntry implementation. Cross-method calls return through the stable SendChainService seam. */ export class SendInboundEntryService { private readonly logger = new Logger('SendChainService'); + private readonly inboundWorkflowWorkerId = `${hostname()}:${process.pid}:${randomUUID()}`; + private readonly inboundWorkflowTasks = new Set>(); + private inboundWorkflowTimer?: ReturnType; + private inboundWorkflowInFlight = 0; + private inboundWorkflowPumping = false; + private inboundWorkflowStopping = false; + private inboundWorkflowMetricsUpdatedAt = 0; constructor( private readonly prisma: PrismaService, @@ -201,6 +224,26 @@ async submitInboundMessage(data: GatewayInboundSubmitDto) { }; } try { + if (this.inboundFastPathEnabled()) { + const response = await this.measureInboundStage('inbox_persist', () => ( + this.persistInboundWorkflow({ + ...data, + content: collection.content, + sequenceId: collection.sequenceId, + registeredDelivery: collection.registeredDelivery ? 1 : 0, + longMessage: undefined, + }, phoneNumbers, application, collection.messageId) + )); + await this.prisma.cmppInboundLongMessage.update({ + where: { id: collection.groupId }, + data: { + status: 'completed', + response: JSON.parse(JSON.stringify(response)) as Prisma.InputJsonValue, + completedAt: new Date(), + }, + }); + return response; + } const response = await this.measureInboundStage('complete_submit', async () => ( await this.facade.recoverCompletedInboundLongMessageResponse( collection.messageId, @@ -233,12 +276,90 @@ async submitInboundMessage(data: GatewayInboundSubmitDto) { throw error; } } + if (this.inboundFastPathEnabled()) { + return this.measureInboundStage( + 'inbox_persist', + () => this.persistInboundWorkflow(data, phoneNumbers, application), + ); + } return this.measureInboundStage( 'complete_submit', () => this.facade.submitCompleteInboundMessage(data, phoneNumbers, application), ); } + private inboundFastPathEnabled() { + return process.env.CMPP_INBOUND_FAST_PATH_ENABLED === 'true'; + } + + private async persistInboundWorkflow( + data: GatewayInboundSubmitDto, + phoneNumbers: string[], + application: NonNullable>>, + requestedGroupMessageId?: string, + ) { + const requestKey = data.requestId?.trim(); + if (!requestKey || requestKey.length > 160) { + throw new BadRequestException('CMPP inbound requestId is required for fast-path idempotency'); + } + if (data.remoteIp && !isIpAllowed(data.remoteIp, application.ipAllowlist.map((item) => item.ipCidr))) { + throw new BadRequestException('CMPP source IP is not in application allowlist'); + } + validateInboundApplicationSrcId(data.srcId, application); + const submitGroupMessageId = requestedGroupMessageId ?? `MSG-${randomUUID()}`; + const messageIds = phoneNumbers.map((_, index) => index === 0 ? submitGroupMessageId : `MSG-${randomUUID()}`); + const payload: InboundWorkflowPayload = { + data: JSON.parse(JSON.stringify(data)) as GatewayInboundSubmitDto, + phoneNumbers, + submitGroupMessageId, + messageIds, + }; + const payloadJson = JSON.parse(JSON.stringify(payload)) as Prisma.InputJsonValue; + const payloadHash = createHash('sha256').update(JSON.stringify({ + data: payload.data, + phoneNumbers, + requestedGroupMessageId: requestedGroupMessageId ?? null, + })).digest('hex'); + const response = { + accepted: true, + tenantId: application.tenantId, + applicationId: application.id, + taskId: '', + messageId: submitGroupMessageId, + messageRecordId: '', + status: 'accepted_pending', + phoneCount: phoneNumbers.length, + messages: phoneNumbers.map((phoneNumber, index) => ({ + phoneNumber, + messageId: messageIds[index], + messageRecordId: '', + taskId: '', + status: 'accepted_pending', + })), + }; + try { + await this.prisma.cmppInboundSubmissionInbox.create({ + data: { + requestKey, + payloadHash, + tenantId: application.tenantId, + applicationId: application.id, + queuePriority: normalizeQueuePriority(application.queuePriority), + payload: payloadJson, + response: JSON.parse(JSON.stringify(response)) as Prisma.InputJsonValue, + }, + }); + return response; + } catch (error) { + if (!(error instanceof Prisma.PrismaClientKnownRequestError) || error.code !== 'P2002') throw error; + const existing = await this.prisma.cmppInboundSubmissionInbox.findUnique({ where: { requestKey } }); + if (!existing || existing.payloadHash !== payloadHash) { + throw new BadRequestException('CMPP inbound requestId conflicts with another payload'); + } + return existing.response as typeof response; + } + } + async recoverCompletedInboundLongMessageResponse(messageId: string, phoneNumbers: string[]) { const existing = await this.prisma.smsMessageRecord.findMany({ where: { @@ -289,6 +410,8 @@ async submitCompleteInboundMessage( phoneNumbers: string[], application: Awaited>, requestedGroupMessageId?: string, + requestedMessageIds?: string[], + workflowKey?: string, ) { if (!application) { throw new BadRequestException('CMPP account is invalid'); @@ -309,6 +432,7 @@ async submitCompleteInboundMessage( phoneNumber: true, status: true, errorCode: true, + batchTask: { select: { status: true } }, }, }) : []; @@ -318,7 +442,11 @@ async submitCompleteInboundMessage( !persistedByPhone.has(phoneNumber) && !phoneRejections.has(phoneNumber) )).length; const dailyQuota = missingPhoneCount > 0 - ? await this.facade.tryReserveDailySendQuota(application.id, missingPhoneCount) + ? await this.facade.tryReserveDailySendQuota( + application.id, + missingPhoneCount, + workflowKey ? `${workflowKey}:daily-quota` : undefined, + ) : { reserved: true, dailyLimit: application.dailyLimit ?? 100000 }; return { persistedByPhone, phoneRejections, dailyQuota, missingPhoneCount }; }); @@ -336,13 +464,19 @@ async submitCompleteInboundMessage( persisted: persistedByPhone.get(phoneNumber), receiptRejection: phoneRejections.get(phoneNumber), messageId: persistedByPhone.get(phoneNumber)?.messageId + ?? requestedMessageIds?.[index] ?? (index === 0 ? submitGroupMessageId : `MSG-${randomUUID()}`), + workflowItemKey: workflowKey ? `${workflowKey}:message:${index}` : undefined, })); const results: GatewayInboundSingleSubmitResult[] = []; const concurrency = 10; for (let offset = 0; offset < submissions.length; offset += concurrency) { const batch = submissions.slice(offset, offset + concurrency); results.push(...await Promise.all(batch.map((submission) => submission.persisted + && !(workflowKey && ( + submission.persisted.status === 'validating' + || (submission.persisted.status === 'queued' && submission.persisted.batchTask?.status !== 'queued') + )) ? Promise.resolve({ accepted: submission.persisted.errorCode !== 'DAILY_LIMIT', tenantId: submission.persisted.tenantId ?? application.tenantId, @@ -356,7 +490,7 @@ async submitCompleteInboundMessage( ...data, phoneNumber: submission.phoneNumber, phoneNumbers: undefined, - }, submission.messageId, submitGroupMessageId, application, submission.receiptRejection ? undefined : dailyLimitRejection, submission.receiptRejection)))); + }, submission.messageId, submitGroupMessageId, application, submission.receiptRejection ? undefined : dailyLimitRejection, submission.receiptRejection, submission.workflowItemKey)))); } const first = results[0]; return { @@ -553,6 +687,7 @@ async submitInboundSingleMessage( application: NonNullable>>, synchronousRejection?: { code: string; reason: string }, receiptRejection?: { code: string; reason: string }, + workflowItemKey?: string, ) { // 入口已按账号取得并校验同一个应用快照；复用它可避免每个目标号码再次查询应用、企业和IP白名单。 if (data.remoteIp && !isIpAllowed(data.remoteIp, application.ipAllowlist.map((item) => item.ipCidr))) { @@ -573,60 +708,101 @@ async submitInboundSingleMessage( phoneCount: 1, unitPrice, }); - const task = await this.measureInboundStage('task_persist', () => this.prisma.smsBatchTask.create({ - data: { - tenantId: application.tenantId, - applicationId: application.id, - templateId: template?.id, - taskNo: `BT-${Date.now()}-${randomUUID().slice(0, 8)}`, - sourceType: 'cmpp', - content: data.content, - phoneTotal: 1, - status: synchronousRejection ? 'rejected' : 'validating', - auditStatus: synchronousRejection ? 'rejected' : undefined, - rejectReason: synchronousRejection?.reason, - progressTotal: 1, - }, - })); - await this.measureInboundStage('api_request_persist', () => this.prisma.smsApiRequest.create({ - data: { - tenantId: application.tenantId, - batchTaskId: task.id, - requestId: `REQ-${Date.now()}-${randomUUID().slice(0, 8)}`, - sourceIp: data.remoteIp, - userAgent: 'cmpp-gateway', - payloadSummary: { phoneTotal: 1, contentLength: [...data.content].length, account: data.account }, - status: synchronousRejection ? 'rejected' : 'accepted', - }, - })); const drainageDetection = await this.measureInboundStage( 'content_detection', () => detectDrainageContent(this.prisma, data.content), ); - const message = await this.measureInboundStage('message_persist', () => this.prisma.smsMessageRecord.create({ - data: { - tenantId: application.tenantId, - batchTaskId: task.id, - applicationId: application.id, - templateId: template?.id, - messageId, - phoneNumber: data.phoneNumber, - content: data.content, - ...drainageDetection, - billingUnits: billing.billingUnitsPerMessage, - unitPrice: receiptRejection ? 0 : billing.unitPrice, - amountCents: receiptRejection ? 0 : billing.amountCents, + const workflowDigest = workflowItemKey + ? createHash('sha256').update(workflowItemKey).digest('hex').slice(0, 32) + : undefined; + let recoveredExisting = false; + let persisted; + try { + persisted = await this.measureInboundStage('message_persist', () => this.prisma.$transaction(async (tx) => { + const task = await tx.smsBatchTask.create({ + data: { + tenantId: application.tenantId, + applicationId: application.id, + templateId: template?.id, + taskNo: workflowDigest ? `BT-IN-${workflowDigest}` : `BT-${Date.now()}-${randomUUID().slice(0, 8)}`, + sourceType: 'cmpp', + content: data.content, + phoneTotal: 1, + status: synchronousRejection ? 'rejected' : 'validating', + auditStatus: synchronousRejection ? 'rejected' : undefined, + rejectReason: synchronousRejection?.reason, + progressTotal: 1, + }, + }); + await tx.smsApiRequest.create({ + data: { + tenantId: application.tenantId, + batchTaskId: task.id, + requestId: workflowDigest ? `REQ-IN-${workflowDigest}` : `REQ-${Date.now()}-${randomUUID().slice(0, 8)}`, + sourceIp: data.remoteIp, + userAgent: 'cmpp-gateway', + payloadSummary: { phoneTotal: 1, contentLength: [...data.content].length, account: data.account }, + status: synchronousRejection ? 'rejected' : 'accepted', + }, + }); + const message = await tx.smsMessageRecord.create({ + data: { + tenantId: application.tenantId, + batchTaskId: task.id, + applicationId: application.id, + templateId: template?.id, + messageId, + phoneNumber: data.phoneNumber, + content: data.content, + ...drainageDetection, + billingUnits: billing.billingUnitsPerMessage, + unitPrice: receiptRejection ? 0 : billing.unitPrice, + amountCents: receiptRejection ? 0 : billing.amountCents, + queuePriority, + cmppSubmitSequenceId: data.sequenceId == null ? null : String(data.sequenceId), + cmppSubmitGroupMessageId: submitGroupMessageId, + cmppRegisteredDelivery: data.registeredDelivery !== 0, + clientSrcId, + applicationExtension: application.cmppApplicationExtension, + status: synchronousRejection ? 'rejected' : 'validating', + errorCode: synchronousRejection?.code, + errorMessage: synchronousRejection?.reason, + }, + }); + return { task, message }; + })); + } catch (error) { + if (!workflowItemKey || !(error instanceof Prisma.PrismaClientKnownRequestError) || error.code !== 'P2002') throw error; + const existing = await this.prisma.smsMessageRecord.findUnique({ + where: { messageId }, + include: { batchTask: true }, + }); + if (!existing?.batchTask || existing.cmppSubmitGroupMessageId !== submitGroupMessageId + || existing.phoneNumber !== data.phoneNumber || existing.applicationId !== application.id) { + throw error; + } + recoveredExisting = true; + persisted = { task: existing.batchTask, message: existing }; + } + const { task, message } = persisted; + + if (recoveredExisting && message.status === 'queued' && task.status !== 'queued') { + await this.facade.enqueueBatchTask(task.id, { + messageRecordId: message.id, queuePriority, - cmppSubmitSequenceId: data.sequenceId == null ? null : String(data.sequenceId), - cmppSubmitGroupMessageId: submitGroupMessageId, - cmppRegisteredDelivery: data.registeredDelivery !== 0, - clientSrcId, - applicationExtension: application.cmppApplicationExtension, - status: synchronousRejection ? 'rejected' : 'validating', - errorCode: synchronousRejection?.code, - errorMessage: synchronousRejection?.reason, - }, - })); + }); + } + if (recoveredExisting && message.status !== 'validating') { + return { + accepted: message.status !== 'rejected' && message.status !== 'failed', + tenantId: application.tenantId, + applicationId: application.id, + taskId: task.id, + messageId: message.messageId, + messageRecordId: message.id, + status: message.status, + }; + } if (synchronousRejection) { return { @@ -658,7 +834,7 @@ async submitInboundSingleMessage( variables: options.templateId ? templateVariables : undefined, phoneNumber: data.phoneNumber, sourceType: 'cmpp', - }); + }, workflowItemKey ? `${workflowItemKey}:frequency` : undefined); return { drainageInfoId: drainage?.id, risk: evaluatedRisk }; }); if (risk.status === 'rejected') { @@ -693,6 +869,7 @@ async submitInboundSingleMessage( relatedType: 'sms_batch_task', relatedId: task.id, remark: 'CMPP 入站短信冻结', + idempotencyKey: workflowItemKey ? `${workflowItemKey}:freeze` : undefined, }); } return check; @@ -736,7 +913,7 @@ async submitInboundSingleMessage( content: data.content, phoneNumber: data.phoneNumber, sourceType: 'cmpp', - }); + }, workflowItemKey ? `${workflowItemKey}:frequency` : undefined); if (risk.status === 'rejected') { await reject('RISK', risk.reason || '短信被风控拒绝'); } else { @@ -754,6 +931,7 @@ async submitInboundSingleMessage( relatedType: 'sms_batch_task', relatedId: task.id, remark: 'CMPP 模板不匹配待审核短信冻结', + idempotencyKey: workflowItemKey ? `${workflowItemKey}:freeze` : undefined, }); } const reviewTask = risk.status === 'pending_review' && risk.task @@ -813,7 +991,7 @@ async evaluateRiskWithPhoneFrequency(input: { variables?: Record; phoneNumber: string; sourceType: 'cmpp'; - }) { + }, reservationKey?: string) { const risk = await this.riskReview.evaluateTask({ tenantId: input.tenantId, applicationId: input.applicationId, @@ -829,6 +1007,8 @@ async evaluateRiskWithPhoneFrequency(input: { input.applicationId, [input.phoneNumber], input.sourceType, + new Date(), + reservationKey, ); const rejection = frequencyRejections.get(input.phoneNumber); return rejection @@ -836,6 +1016,170 @@ async evaluateRiskWithPhoneFrequency(input: { : risk; } +startInboundWorkflowWorker() { + if (this.inboundWorkflowTimer || this.inboundWorkflowPumping || this.inboundWorkflowTasks.size > 0) { + return { status: 'already_started' }; + } + this.inboundWorkflowStopping = false; + this.scheduleInboundWorkflowPump(0); + return { status: 'started' }; + } + + async stopInboundWorkflowWorker() { + this.inboundWorkflowStopping = true; + if (this.inboundWorkflowTimer) clearTimeout(this.inboundWorkflowTimer); + this.inboundWorkflowTimer = undefined; + await Promise.allSettled([...this.inboundWorkflowTasks]); + } + + private scheduleInboundWorkflowPump(delayMs: number) { + if (this.inboundWorkflowStopping || this.inboundWorkflowTimer) return; + this.inboundWorkflowTimer = setTimeout(() => { + this.inboundWorkflowTimer = undefined; + void this.pumpInboundWorkflow(); + }, delayMs); + this.inboundWorkflowTimer.unref?.(); + } + + private async pumpInboundWorkflow() { + if (this.inboundWorkflowStopping || this.inboundWorkflowPumping) return; + const concurrency = positiveInteger(process.env.API_INBOUND_WORKFLOW_CONCURRENCY, 32); + this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight); + const available = Math.max(0, concurrency - this.inboundWorkflowInFlight); + if (available === 0) return; + this.inboundWorkflowPumping = true; + try { + await this.refreshInboundWorkflowMetrics(); + const claimed = await this.claimInboundWorkflows(available); + for (const item of claimed) { + this.inboundWorkflowInFlight += 1; + this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight); + const task = this.processClaimedInboundWorkflow(item) + .catch((error) => this.logger.error(`CMPP inbound workflow ${item.id} failed to settle: ${String(error)}`)) + .finally(() => { + this.inboundWorkflowInFlight = Math.max(0, this.inboundWorkflowInFlight - 1); + this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight); + this.inboundWorkflowTasks.delete(task); + this.scheduleInboundWorkflowPump(0); + }); + this.inboundWorkflowTasks.add(task); + } + if (claimed.length === 0) { + this.scheduleInboundWorkflowPump(positiveInteger(process.env.API_INBOUND_WORKFLOW_POLL_INTERVAL_MS, 100)); + } + } catch (error) { + this.logger.error(`Failed to claim CMPP inbound workflow: ${String(error)}`); + this.scheduleInboundWorkflowPump(1000); + } finally { + this.inboundWorkflowPumping = false; + if (this.inboundWorkflowInFlight < concurrency && !this.inboundWorkflowTimer) { + this.scheduleInboundWorkflowPump(0); + } + } + } + + private claimInboundWorkflows(limit: number) { + const staleSeconds = positiveInteger(process.env.API_INBOUND_WORKFLOW_STALE_SECONDS, 300); + return this.prisma.$queryRaw(Prisma.sql` + WITH candidates AS ( + SELECT id + FROM "CmppInboundSubmissionInbox" + WHERE ( + status = 'pending' + AND "nextAttemptAt" <= NOW() + ) OR ( + status = 'processing' + AND "lockedAt" <= NOW() - make_interval(secs => ${staleSeconds}) + ) + -- 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. + ORDER BY CASE WHEN "queuePriority" = 'priority' THEN 0 ELSE 1 END, "createdAt" ASC + LIMIT ${limit} + FOR UPDATE SKIP LOCKED + ) + UPDATE "CmppInboundSubmissionInbox" AS inbox + SET status = 'processing', + attempts = inbox.attempts + 1, + "lockedAt" = NOW(), + "lockedBy" = ${this.inboundWorkflowWorkerId}, + "updatedAt" = NOW() + FROM candidates + WHERE inbox.id = candidates.id + RETURNING inbox.id, inbox."requestKey", inbox."applicationId", inbox.attempts, inbox.payload + `); + } + + private async processClaimedInboundWorkflow(item: ClaimedInboundWorkflow) { + try { + const payload = parseInboundWorkflowPayload(item.payload); + const application = await this.facade.findInboundApplication(payload.data.account); + if (!application || application.id !== item.applicationId) { + throw new Error('CMPP inbound application no longer matches persisted workflow'); + } + const result = await this.facade.submitCompleteInboundMessage( + payload.data, + payload.phoneNumbers, + application, + payload.submitGroupMessageId, + payload.messageIds, + item.requestKey, + ); + const settled = await this.prisma.cmppInboundSubmissionInbox.updateMany({ + where: { id: item.id, status: 'processing', lockedBy: this.inboundWorkflowWorkerId }, + data: { + status: 'completed', + result: JSON.parse(JSON.stringify(result)) as Prisma.InputJsonValue, + completedAt: new Date(), + lockedAt: null, + lockedBy: null, + lastError: null, + }, + }); + if (settled.count !== 1) throw new Error('CMPP inbound workflow lease was lost before completion'); + this.metrics?.recordInboundWorkflowResult('completed'); + } catch (error) { + const reason = (error instanceof Error ? error.message : String(error)).slice(0, 2000); + const delayMs = Math.min(60_000, 250 * 2 ** Math.min(8, Math.max(0, item.attempts - 1))); + const released = await this.prisma.cmppInboundSubmissionInbox.updateMany({ + where: { id: item.id, status: 'processing', lockedBy: this.inboundWorkflowWorkerId }, + data: { + status: 'pending', + nextAttemptAt: new Date(Date.now() + delayMs), + lockedAt: null, + lockedBy: null, + lastError: reason, + }, + }); + if (released.count === 1) { + this.metrics?.recordInboundWorkflowResult('retry'); + this.logger.warn(`CMPP inbound workflow ${item.id} will retry after attempt ${item.attempts}: ${reason}`); + return; + } + throw error; + } + } + + private async refreshInboundWorkflowMetrics() { + const now = Date.now(); + if (now - this.inboundWorkflowMetricsUpdatedAt < 5_000) return; + this.inboundWorkflowMetricsUpdatedAt = now; + const [pending, processing, oldest] = await Promise.all([ + this.prisma.cmppInboundSubmissionInbox.count({ where: { status: 'pending' } }), + this.prisma.cmppInboundSubmissionInbox.count({ where: { status: 'processing' } }), + this.prisma.cmppInboundSubmissionInbox.findFirst({ + where: { status: 'pending' }, + select: { createdAt: true }, + orderBy: { createdAt: 'asc' }, + }), + ]); + this.metrics?.setInboundWorkflowState( + pending, + processing, + oldest ? Math.max(0, (now - oldest.createdAt.getTime()) / 1000) : 0, + ); + } + findInboundApplication(account: string) { return this.prisma.smsApplication.findFirst({ where: { cmppAccount: account }, @@ -917,3 +1261,24 @@ async attachMessageToReviewTask(reviewTaskId: string, messageRecordId: string, s return this.prisma.smsSendTask.findUnique({ where: { id: reviewTaskId } }); } } + +function parseInboundWorkflowPayload(value: Prisma.JsonValue): InboundWorkflowPayload { + if (!value || typeof value !== 'object' || Array.isArray(value)) throw new Error('CMPP inbound workflow payload is invalid'); + const data = value.data; + const phoneNumbers = value.phoneNumbers; + const submitGroupMessageId = value.submitGroupMessageId; + const messageIds = value.messageIds; + if (!data || typeof data !== 'object' || Array.isArray(data) + || !Array.isArray(phoneNumbers) || phoneNumbers.some((item) => typeof item !== 'string') + || typeof submitGroupMessageId !== 'string' + || !Array.isArray(messageIds) || messageIds.some((item) => typeof item !== 'string') + || phoneNumbers.length === 0 || phoneNumbers.length !== messageIds.length) { + throw new Error('CMPP inbound workflow payload fields are invalid'); + } + return { + data: data as unknown as GatewayInboundSubmitDto, + phoneNumbers: phoneNumbers as string[], + submitGroupMessageId, + messageIds: messageIds as string[], + }; +} diff --git a/api/src/send-chain/send-submission.service.ts b/api/src/send-chain/send-submission.service.ts index 4e63c5e..96a296d 100644 --- a/api/src/send-chain/send-submission.service.ts +++ b/api/src/send-chain/send-submission.service.ts @@ -67,7 +67,10 @@ export class SendSubmissionService { } onModuleDestroy() { - return this.gatewaySubmit.onModuleDestroy(); + return Promise.all([ + this.gatewaySubmit.onModuleDestroy(), + this.inboundEntry.stopInboundWorkflowWorker(), + ]); } async createBatchTask(data: CreateBatchTaskDto) { @@ -123,8 +126,8 @@ async reserveDailySendQuota(applicationId: string, requestedCount: number) { return this.batchEntry.reserveDailySendQuota(applicationId, requestedCount); } -async tryReserveDailySendQuota(applicationId: string, requestedCount: number) { - return this.batchEntry.tryReserveDailySendQuota(applicationId, requestedCount); +async tryReserveDailySendQuota(applicationId: string, requestedCount: number, reservationKey?: string) { + return this.batchEntry.tryReserveDailySendQuota(applicationId, requestedCount, reservationKey); } async authenticateInboundApplication(data: GatewayInboundAuthDto) { @@ -144,8 +147,10 @@ async submitCompleteInboundMessage( phoneNumbers: string[], application: Awaited>, requestedGroupMessageId?: string, + requestedMessageIds?: string[], + workflowKey?: string, ) { - return this.inboundEntry.submitCompleteInboundMessage(data, phoneNumbers, application, requestedGroupMessageId); + return this.inboundEntry.submitCompleteInboundMessage(data, phoneNumbers, application, requestedGroupMessageId, requestedMessageIds, workflowKey); } async collectInboundLongMessageFragment( @@ -167,8 +172,9 @@ async submitInboundSingleMessage( application: NonNullable>>, synchronousRejection?: { code: string; reason: string }, receiptRejection?: { code: string; reason: string }, + workflowItemKey?: string, ) { - return this.inboundEntry.submitInboundSingleMessage(data, messageId, submitGroupMessageId, application, synchronousRejection, receiptRejection); + return this.inboundEntry.submitInboundSingleMessage(data, messageId, submitGroupMessageId, application, synchronousRejection, receiptRejection, workflowItemKey); } async evaluateRiskWithPhoneFrequency(input: { @@ -179,8 +185,8 @@ async evaluateRiskWithPhoneFrequency(input: { variables?: Record; phoneNumber: string; sourceType: 'cmpp'; - }) { - return this.inboundEntry.evaluateRiskWithPhoneFrequency(input); + }, reservationKey?: string) { + return this.inboundEntry.evaluateRiskWithPhoneFrequency(input, reservationKey); } findInboundApplication(account: string) { @@ -223,6 +229,14 @@ startWorker() { return this.gatewaySubmit.startWorker(); } +startInboundWorkflowWorker() { + return this.inboundEntry.startInboundWorkflowWorker(); + } + +stopInboundWorkflowWorker() { + return this.inboundEntry.stopInboundWorkflowWorker(); + } + async processSendJob(job: SendJob) { return this.gatewaySubmit.processSendJob(job); } diff --git a/api/src/send-worker.module.ts b/api/src/send-worker.module.ts new file mode 100644 index 0000000..346329b --- /dev/null +++ b/api/src/send-worker.module.ts @@ -0,0 +1,20 @@ +import { Module } from '@nestjs/common'; +import { ConfigModule } from '@nestjs/config'; +import { BillingModule } from './billing/billing.module'; +import { PhoneRoutingLookupService } from './dictionaries/phone-routing-lookup.service'; +import { MetricsModule } from './metrics/metrics.module'; +import { PrismaModule } from './prisma/prisma.module'; +import { PhoneFrequencyService } from './risk-review/phone-frequency.service'; +import { RiskReviewService } from './risk-review/risk-review.service'; +import { SendChainService } from './send-chain/send-chain.service'; + +@Module({ + imports: [ + ConfigModule.forRoot({ isGlobal: true, envFilePath: ['.env.local', '.env'] }), + PrismaModule, + BillingModule, + MetricsModule, + ], + providers: [RiskReviewService, PhoneFrequencyService, PhoneRoutingLookupService, SendChainService], +}) +export class SendWorkerModule {} diff --git a/api/src/send-worker.ts b/api/src/send-worker.ts new file mode 100644 index 0000000..5f54bdf --- /dev/null +++ b/api/src/send-worker.ts @@ -0,0 +1,45 @@ +import 'reflect-metadata'; +import { NestFactory } from '@nestjs/core'; +import { createServer } from 'node:http'; +import { MetricsService } from './metrics/metrics.service'; +import { SendWorkerModule } from './send-worker.module'; + +Object.defineProperty(BigInt.prototype, 'toJSON', { + configurable: true, + value(this: bigint) { + const result = Number(this); + if (!Number.isSafeInteger(result)) throw new RangeError('金额超过 JavaScript 安全整数范围'); + return result; + }, +}); + +async function bootstrap() { + if ((process.env.CMPP_PROCESS_ROLE?.trim() || 'worker') !== 'worker') { + throw new Error('send-worker requires CMPP_PROCESS_ROLE=worker'); + } + const app = await NestFactory.createApplicationContext(SendWorkerModule); + app.enableShutdownHooks(); + const metrics = app.get(MetricsService); + const host = process.env.API_WORKER_METRICS_HOST?.trim() || '127.0.0.1'; + const port = Number(process.env.API_WORKER_METRICS_PORT ?? 9465); + const metricsServer = createServer((request, response) => { + if (request.method !== 'GET' || request.url !== '/metrics') { + response.writeHead(404).end(); + return; + } + response.writeHead(200, { 'Content-Type': 'text/plain; version=0.0.4; charset=utf-8', 'Cache-Control': 'no-store' }); + response.end(metrics.render()); + }); + await new Promise((resolve, reject) => { + metricsServer.once('error', reject); + metricsServer.listen(port, host, resolve); + }); + const close = async () => { + await new Promise((resolve) => metricsServer.close(() => resolve())); + await app.close(); + }; + process.once('SIGTERM', () => void close()); + process.once('SIGINT', () => void close()); +} + +void bootstrap(); diff --git a/docs/codebase-modularization-roadmap.md b/docs/codebase-modularization-roadmap.md index ee4f7be..51ba58c 100644 --- a/docs/codebase-modularization-roadmap.md +++ b/docs/codebase-modularization-roadmap.md @@ -1155,5 +1155,7 @@ global，不能错误归入client。`client-signature-*`、发送页、企业认 - V3客户入站窗口只归`gateway/internal/inbound/`与项目内受控的`third_party/gocmpp`服务循环治理：API认证只返回应用窗口，inbound会话负责收紧窗口，协议服务循环负责受限派发和断线等待；不得把客户入站槽位与`submitworker`供应商槽位或`upstream`供应商窗口合并成同一并发计数。 - V4供应商结果异步边界只归`gateway/internal/resultoutbox/`治理：`upstream`在每个真实分片SubmitResp后只调用持久化接口，`submitworker`只负责聚合结果入Outbox与命令ACK的原子边界，Outbox回调Worker独立控制API并发和PEL恢复。API的`SmsSubmitRecord.resultEventId`是跨重启持久幂等事实；不得把HTTP回调重新放回供应商连接池或Submit工作槽，也不得让Outbox承担业务计费、补发或状态机判断。 - V5 API数据库往返优化仍归`send-inbound-entry`、`send-gateway-submit`和`risk-review`现有边界：入口应用快照沿稳定Facade显式传递，单条快速入队只接受已持久化ID和优先级，通用批量入队保留原查询与取消校验；默认规则并发单飞属于`RiskReviewService`内部完整性保障，不得在`SendChainService`新增第二套规则缓存，也不得把余额、频控状态或实际规则决策缓存进进程内存。 +- 500条/秒第一阶段把接收与业务处理边界固定在`CmppInboundSubmissionInbox`：`gateway/internal/inbound`只生成连接域内稳定请求键，`send-inbound-entry`只负责最小校验、Inbox持久化和稳定响应；`api/src/send-worker.ts`在独立进程内领取并调用既有发送链。领取事务不得包住风控、计费、路由、Redis或供应商调用，业务幂等事实必须落PostgreSQL，不能依赖进程内缓存或localStorage。 +- API进程角色固定为`CMPP_PROCESS_ROLE=api`，不启动发送Worker、Inbox Worker和周期扫描；`cmpp-send-worker`固定为`worker`并拥有自己的Prisma连接池和回环指标。后续扩容允许水平增加Worker实例，但不得复制HTTP控制器或绕过Inbox直接创建业务消息。 - `api/src/infrastructure-monitoring/`是运营端监控聚合与固定阈值应用边界：只消费代码白名单 PromQL，并通过版本化 PostgreSQL 单例、promtool 校验和原子规则热加载管理数值阈值；Exporter安装、端口隔离、固定规则模板和权限仍归`tools/monitoring/`治理。 - 活动告警已读也归该边界：Prometheus保留告警事实，Prisma仅持久化逐管理员、逐触发周期的阅读状态；全局布局只消费轻量未读汇总，不复制指纹、activeAt或用户隔离逻辑。 diff --git a/docs/contracts/inbound-r6-declarations.json b/docs/contracts/inbound-r6-declarations.json index 1583efc..d71980b 100644 --- a/docs/contracts/inbound-r6-declarations.json +++ b/docs/contracts/inbound-r6-declarations.json @@ -516,7 +516,7 @@ "name": "handleSubmit", "kind": "func", "file": "submit.go", - "sha256": "16a7bb939824b18ea13698293f2868a25a7cb525e318a1c6b00597f37be12466" + "sha256": "36b9016de0d3cf0d2286b00aad90435d67cc3720ba5fcc87a3d27e8c3db4d2c1" }, { "name": "inboundLongMessageFragment", @@ -530,6 +530,12 @@ "file": "submit.go", "sha256": "adac6aed04068f3811fbcf0530ab54f7384f0ea85a303c9c70326b5a9a46ba78" }, + { + "name": "inboundSubmitRequestID", + "kind": "func", + "file": "submit.go", + "sha256": "5f3515253ece00423b96d0c8b32efdefa4b4c61d88991997df66c9acccce4b7e" + }, { "name": "messageIDFrom", "kind": "func", @@ -564,7 +570,7 @@ "name": "submitRequest", "kind": "type", "file": "submit.go", - "sha256": "80791da6ef4e968dbafbea288ef783422b4c089013a05175b6add32bf8919df0" + "sha256": "7edc6ef9b65d44dd3c2f22415c3d6ceca12621cbf531b089fb500ecfea282253" }, { "name": "submitResponse", diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index dfc9b02..d000a9a 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -2103,6 +2103,10 @@ - V3入站并发必须只并发Submit业务处理，连接认证保持串行先完成，心跳和Deliver ACK不得被长耗时Submit阻塞；每个SubmitResp继续使用原请求Sequence_Id关联，允许按实际完成顺序返回。同一连接关闭时必须先等待已接受的在途处理收尾，再清理会话和回执映射，避免迟到处理重新注册已断开的连接。`cmpp_gateway_inbound_submit_slots{state=configured|in_flight}`只暴露全部在线连接的聚合窗口与在途数量，不得增加账号、应用、连接或消息标签。 - V4结果Outbox必须暴露独立回调Worker的configured/in_flight槽位和Outbox pending/lag，仍只使用固定状态标签。`api_callback`从V4起只在Outbox回调Worker计时，不再混入供应商Submit工作槽；压测结束必须同时核对命令Stream和结果Outbox均`pending=0/lag=0`，并证明API回调故障时供应商槽继续释放、结果不丢失且恢复后只处理一次。 - V5 API入站数据库往返治理不得改变SubmitResp、余额冻结、号码频控、模板/签名、路由、队列或失败回执语义。同一Submit入口已取得的应用、企业和IP白名单快照必须复用于其全部目标号码，不得逐号码重复查询；任务风控与号码频控共用的默认规则完整性检查必须使用短TTL并发单飞，正常完整状态最多执行一次聚合数据库检查，实际生效规则仍逐次读取；刚持久化的单条CMPP内部任务和消息可凭已知ID、优先级直接入队，不得为入队重新查询同一任务和消息。缓存只覆盖默认规则“是否齐全”，检查失败必须立即失效，规则缺失须在短TTL到期后自动恢复；不得缓存余额、频控计数、应用启停或实际生效规则。 +- 面向完整处理500条/秒目标的第一阶段，CMPP SubmitResp语义调整为“已完成最小协议/账号/IP/Src_Id校验且已持久化接收事实”，不再同步等待模板、风控、频控、日限、计费、路由和入队。Gateway必须为同一连接内同一Submit重试生成稳定请求键；PostgreSQL Inbox以唯一请求键和载荷摘要幂等保存原请求、稳定内部MessageId及响应。相同键相同载荷返回原响应，相同键不同载荷必须拒绝。 +- Inbox Worker必须作为独立非root进程运行，使用独立数据库连接池和持续有界工作池；领取使用短事务、`FOR UPDATE SKIP LOCKED`和过期租约回收，实际业务处理在领取事务外执行。成功逐条完成，失败以有上限退避回到pending且不得静默丢弃。日发送配额、号码频控及余额冻结必须使用由Inbox请求派生的持久幂等键，确保进程崩溃或租约回收不会重复扣量、重复计频或重复冻结。 +- 优先应用和普通应用共享同一耐久Inbox，但领取顺序必须保留优先级并在同一优先级内FIFO；进入BullMQ后继续沿用priority=1、normal=100。快路径成功只代表平台已可靠接收，异步业务拒绝仍必须落真实消息/任务状态并按既有CMPP失败回执链路通知客户，不得伪装为供应商最终送达。 +- 第一阶段的验收是入口可持续接收、Inbox不丢不重且最终可排空，并为后续500条/秒全链路扩容建立解耦边界；不能仅凭SubmitResp吞吐宣称完整500条/秒。压测必须同时报告SubmitResp成功率/延迟、Inbox pending/processing/最老等待、异步完成速率和排空时间，以及命令Stream、结果Outbox和数据库最终对账。 - 系统监控标题说明必须明确标注数据来自 Prometheus；“服务关键指标”位于趋势/核心服务区域之后、活动告警之前，并提供统一的“告警阈值设置”入口。安全检测页不重复渲染大号标题，说明文字必须明确标注使用 Fail2ban。 - 告警阈值仅开放固定指标的警告/严重数值，不允许前端提交 PromQL、标签、文件路径或持续时间；必须满足警告值小于严重值。配置以 PostgreSQL 保存版本、期望值、生效值和应用状态，经 `promtool check rules` 校验、同目录原子替换和 Prometheus 热加载成功后才标记生效，失败保留上一生效规则并展示原因。 - 右上角预警中心增加“系统监控告警”，通过独立轻量接口统计 Prometheus 当前 firing/pending 告警及严重数，跳转系统监控活动告警区；任一预警域失败不得清空其他域。 diff --git a/docs/production-deployment.md b/docs/production-deployment.md index 393217c..fc125ba 100644 --- a/docs/production-deployment.md +++ b/docs/production-deployment.md @@ -152,6 +152,9 @@ 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_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:*' redis-cli XINFO GROUPS gateway.submit.commands redis-cli XINFO GROUPS gateway.submit.results @@ -174,3 +177,10 @@ bash tools/deploy/production-deploy.sh ``` 3. 如迁移造成不可兼容故障，先停服务，再恢复数据库备份。 + +## CMPP耐久Inbox与独立Worker发布门禁（2026-08-20） + +- 发布前恢复资产除PostgreSQL、运行源码和环境文件外，必须包含`cmpp-api.service`、`cmpp-send-worker.service`及其drop-in；逐项校验`pg_restore --list`、tar可读性和SHA-256。数据库回滚与运行代码必须成套执行，禁止只回退代码后让旧Prisma Client访问新状态机。 +- 环境必须显式启用`CMPP_INBOUND_FAST_PATH_ENABLED=true`、`CMPP_INBOUND_WORKFLOW_WORKER_ENABLED=true`，并给出正整数`API_INBOUND_WORKFLOW_CONCURRENCY`；推荐初始值32、轮询100ms、租约300秒。API systemd角色必须是`api`，Worker角色必须是`worker`；Worker可通过`API_WORKER_DATABASE_URL`使用独立连接上限，未配置时仍使用同一数据库地址但保持独立进程连接池。 +- 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或重投真实短信。 diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index f6ddcd0..e913b65 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -4726,3 +4726,18 @@ npm run verify:phase8 | TC-INFRA-MON-037 | 已读用户隔离 | 管理员A标记已读后由管理员B查看同一告警 | 管理员B仍显示未读且铃铛数量不减少，管理员A的状态保持已读 | | TC-INFRA-MON-038 | 同告警重新触发 | 标记已读后让告警恢复，再以相同标签重新触发并产生新 activeAt | 新触发记录重新显示“标记已读”，计入预警中心；旧 activeAt 不会永久屏蔽同指纹告警 | | TC-INFRA-MON-039 | 过期与幂等 | 重复提交同一活动告警，再提交已恢复或 activeAt 不匹配的请求 | 同一次告警重复提交幂等；过期/不匹配请求返回404且不生成虚假已读记录；操作日志可追溯 | + +## CMPP 500条/秒第一阶段：耐久Inbox快路径（2026-08-20） + +| 用例ID | 场景 | 步骤 | 预期 | +| --- | --- | --- | --- | +| TC-CMPP-500-P1-001 | 快速耐久受理 | 开启快路径提交合法短消息，并在风控/计费/队列依赖可观测时检查调用顺序 | SubmitResp在一条Inbox事实提交后返回`accepted_pending`；响应前不调用风控、频控、计费或BullMQ | +| TC-CMPP-500-P1-002 | 请求幂等 | 在同一已鉴权连接重投相同Sequence_Id与载荷 | Gateway请求键稳定，数据库只保留一条Inbox；两次返回相同MessageId，不重复创建任务、消息或冻结 | +| TC-CMPP-500-P1-003 | 幂等冲突 | 使用同一请求键提交不同载荷 | API拒绝冲突且不覆盖原Inbox/响应 | +| TC-CMPP-500-P1-004 | 多号码与长短信 | 分别提交多号码Submit和完整长短信分片 | Inbox保留全部号码及稳定子MessageId；长短信Worker处理的是完整重组正文，不是最后一片正文 | +| TC-CMPP-500-P1-005 | Worker领取与逐条完成 | 启动两个Worker并制造至少一个批次积压 | `SKIP LOCKED`领取不重复；每条独立完成，处理逻辑不占用领取事务 | +| TC-CMPP-500-P1-006 | 崩溃恢复 | 在领取后终止Worker，超过租约后重启 | processing记录被回收并完成；日限、频控、冻结、任务和消息均不重复 | +| TC-CMPP-500-P1-007 | 异步业务拒绝 | 让已耐久受理短信命中真实模板/风控/余额拒绝 | SubmitResp仍表示已接收；后台生成真实失败状态和失败回执，不进入供应商发送 | +| TC-CMPP-500-P1-008 | 优先级 | 在普通Inbox积压期间持续混入priority应用 | priority先领取且类内FIFO；普通队列最终可排空；同时记录两类等待分位数 | +| TC-CMPP-500-P1-009 | 进程隔离 | 检查systemd、进程、连接和指标端口 | API角色不运行发送后台任务；`cmpp-send-worker`独立非root运行，指标仅监听127.0.0.1:9465 | +| TC-CMPP-500-P1-010 | 阶梯压测与对账 | 隔离供应商环境按既定同口径阶梯执行，压后等待全链排空 | 逐档报告SubmitResp、Inbox、两条Stream、最终数据库计数和排空时间；任何丢响应、重复、错误或未排空均失败，不发送真实短信 | diff --git a/docs/testing-progress.md b/docs/testing-progress.md index 3facafa..b3b8da6 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -3782,3 +3782,12 @@ git diff --check - Prometheus按各60秒窗口附近75秒区间计算的API阶段平均值显示，10/20/30/40/50档`application_lookup`约为`1.42/51.04/65.54/108.64/247.65ms`，`risk_frequency`约`4.34/115.10/143.61/229.05/519.32ms`，`queue_publish`约`4.71/70.79/101.17/163.46/383.60ms`，`complete_submit`约`18.99/479.58/610.03/992.49/2285.77ms`。V5减少固定数据库往返后40条/秒P95继续下降，但50条/秒时同步风控、入队和持久化仍随数据库并发竞争放大，下一阶段应针对这些阶段继续做查询合并/事务缩短，而不是把50条/秒宣称为安全容量。 - 优先队列在真实积压下通过验收。BullMQ配置为priority=1、normal=100；30条/秒时优先任务`queuedAt`至首条`SmsSubmitRecord.createdAt`的P50/P95=`0.786/0.988s`，普通任务为`14.678/26.282s`；40条/秒分别为`2.996/3.916s`与`45.017/58.536s`。低负载10/20条/秒两类接近，符合无积压时无需插队的预期；高负载差异证明优先任务能够越过普通积压持续取得发送Worker。SubmitResp按优先/普通拆分在40条/秒P95为`1727/1741ms`，说明上游受理没有饿死普通连接。50条/秒优先任务也优于普通任务，但两类均出现几十秒下游等待，不能用优先级掩盖总容量过载。 - 本轮优先级结论只覆盖3个priority应用与7个normal应用混合、移动号段和现有六个隔离供应商账号；联通、电信号段及六通道跨运营商容量/优先级隔离仍未完成，不外推为全运营商结论。完整报告和原始`events.jsonl/summary.json`保存在短信平台测试项目。没有发送真实短信，没有修改通道账号、密码、启停状态、企业余额或客户连接；受保护的`api/tsconfig.build.tsbuildinfo`、`tsconfig.tsbuildinfo`、`outputs/`、`pnpm-lock.yaml`和空文件`=`继续排除提交、不删除、不归因。 + +# 2026-08-20 完整处理500条/秒第一阶段：PostgreSQL耐久Inbox快路径（提交前验证） + +- 第一阶段把CMPP Submit同步路径收敛为账号/状态/IP/Src_Id/协议最小校验和一条`CmppInboundSubmissionInbox`持久化；Gateway按已鉴权连接、Sequence_Id和载荷生成稳定请求键。重复相同Submit返回原持久化MessageId，相同请求键的冲突载荷拒绝。多号码保留稳定子MessageId，长短信在完整重组后才写Inbox，Worker处理完整正文。 +- 独立`cmpp-send-worker`进程持续以默认32槽领取Inbox；短事务使用`FOR UPDATE SKIP LOCKED`和300秒过期租约，处理在领取事务外完成，失败按有上限指数退避回到pending。priority应用在领取阶段优先且类内FIFO，BullMQ继续使用priority=1、normal=100。API角色不再运行发送Worker或周期扫描；Worker拥有独立Prisma连接池入口和127.0.0.1:9465指标。 +- 新增持久幂等预留：日发送配额与`SmsApplicationDailyUsage`同事务写`SmsApplicationDailyReservation`，号码频控计数与`PhoneFrequencyReservation`同事务提交，余额冻结继续使用业务键；任务、API请求和短信主记录合并为一个短事务并使用确定性任务号/请求号。租约回收或进程重启不会重复计量、计频、冻结或创建业务主记录。 +- Prisma新增第92条migration `20260820170000_add_cmpp_inbound_submission_inbox`，包含Inbox、日配额预留和号码频控预留三张真实PostgreSQL表及领取/租约/应用索引。发布脚本增加快路径/Worker强制门禁、systemd分进程、安全drop-in、Worker日志目录、健康检查和Prometheus target/告警；需求、系统用例、模块路线图和生产发布文档同步更新。 +- 提交前验证：Prisma generate/validate、API TypeScript正式构建、前端TypeScript与Vite生产构建通过；API全量42套489项通过，Gateway全量`go test ./... -count=1`和`go vet ./...`通过，5份Stream契约、依赖/安全/部署门禁、R0/R6/R7/R10及`git diff --check`通过。R6契约只新增/更新本轮`submit.go`的稳定请求键声明。R9仍先被HEAD既有`dispatchDueScheduledTasks`哈希漂移阻断，与本轮文件和既有V5记录一致，未为通过本任务错误吸收该并行历史。Jest仍用`--forceExit`收尾仓库既有开放句柄，Vite仅保留既有大chunk告警。 +- 当前尚未提交、部署或压测；下一步在排除受保护文件后提交，随后仅对`100.93.204.60`测试环境建立并校验PostgreSQL、运行源码、环境/systemd恢复资产，应用migration和独立Worker，再用隔离供应商执行同口径阶梯压测。预生产不发布、不回退、不压测；不发送、补发或重投真实短信，不修改真实通道账号、密码、启停状态、企业余额或客户连接。 diff --git a/gateway/internal/inbound/submit.go b/gateway/internal/inbound/submit.go index bb7aaa8..13d13bc 100644 --- a/gateway/internal/inbound/submit.go +++ b/gateway/internal/inbound/submit.go @@ -4,6 +4,7 @@ import ( "cmpp-platform/gateway/internal/metrics" "context" "crypto/md5" + "crypto/sha256" "errors" "fmt" cmpp "github.com/bigwhite/gocmpp" @@ -19,6 +20,7 @@ import ( // one internal message mapping per destination while CMPP receives one response. type submitRequest struct { + RequestID string `json:"requestId,omitempty"` Account string `json:"account"` PhoneNumber string `json:"phoneNumber,omitempty"` PhoneNumbers []string `json:"phoneNumbers,omitempty"` @@ -117,6 +119,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge releaseSubmitBarrier := beginDownstreamSubmitBarrier(packet.Conn) apiStartedAt := time.Now() result, err := s.submit(remote, submitRequest{ + RequestID: inboundSubmitRequestID(session.connectionID, req.sequenceID, phones, content, req.srcID, longMessage), Account: account, PhoneNumber: phone, PhoneNumbers: phones, @@ -222,6 +225,14 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge return false, nil } +func inboundSubmitRequestID(connectionID string, sequenceID uint32, phones []string, content string, srcID string, longMessage *inboundLongMessageFragment) string { + // The key is stable for an API retry of the same packet but scoped to the authenticated + // connection, so a later client session may intentionally reuse the CMPP Sequence_Id. + payload := fmt.Sprintf("%s\x00%d\x00%s\x00%s\x00%s\x00%v", connectionID, sequenceID, strings.Join(phones, ","), strings.TrimSpace(srcID), content, longMessage) + digest := sha256.Sum256([]byte(payload)) + return fmt.Sprintf("cmpp-inbound:%x", digest[:]) +} + func observeInboundSubmitResponse(handlerStartedAt time.Time, responseReadyAt time.Time, accepted bool, next func(error)) func(error) { return func(sendErr error) { metrics.ObserveInboundStage("response_write", sendErr == nil, time.Since(responseReadyAt)) diff --git a/gateway/internal/inbound/submit_idempotency_test.go b/gateway/internal/inbound/submit_idempotency_test.go new file mode 100644 index 0000000..9132acc --- /dev/null +++ b/gateway/internal/inbound/submit_idempotency_test.go @@ -0,0 +1,27 @@ +package inbound + +import "testing" + +func TestInboundSubmitRequestIDIsStableForRetry(t *testing.T) { + fragment := &inboundLongMessageFragment{Reference: 7, Total: 2, Index: 1, Format: 8} + first := inboundSubmitRequestID("connection-1", 42, []string{"13800000001"}, "【测试】内容", "1069", fragment) + second := inboundSubmitRequestID("connection-1", 42, []string{"13800000001"}, "【测试】内容", "1069", fragment) + if first != second { + t.Fatalf("retry key changed: %q != %q", first, second) + } +} + +func TestInboundSubmitRequestIDSeparatesPayloadAndSession(t *testing.T) { + base := inboundSubmitRequestID("connection-1", 42, []string{"13800000001"}, "content", "1069", nil) + cases := []string{ + inboundSubmitRequestID("connection-2", 42, []string{"13800000001"}, "content", "1069", nil), + inboundSubmitRequestID("connection-1", 43, []string{"13800000001"}, "content", "1069", nil), + inboundSubmitRequestID("connection-1", 42, []string{"13800000002"}, "content", "1069", nil), + inboundSubmitRequestID("connection-1", 42, []string{"13800000001"}, "other", "1069", nil), + } + for _, candidate := range cases { + if candidate == base { + t.Fatalf("different submit produced duplicate key %q", base) + } + } +} diff --git a/tools/deploy/production-bootstrap.sh b/tools/deploy/production-bootstrap.sh index 3c9d936..8a0d222 100644 --- a/tools/deploy/production-bootstrap.sh +++ b/tools/deploy/production-bootstrap.sh @@ -11,6 +11,13 @@ API_METRICS_HOST="${API_METRICS_HOST:-127.0.0.1}" API_METRICS_PORT="${API_METRICS_PORT:-9464}" API_ENABLE_SEND_WORKER="${API_ENABLE_SEND_WORKER:-true}" API_SEND_WORKER_CONCURRENCY="${API_SEND_WORKER_CONCURRENCY:-50}" +API_WORKER_METRICS_HOST="${API_WORKER_METRICS_HOST:-127.0.0.1}" +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_POLL_INTERVAL_MS="${API_INBOUND_WORKFLOW_POLL_INTERVAL_MS:-100}" +API_INBOUND_WORKFLOW_STALE_SECONDS="${API_INBOUND_WORKFLOW_STALE_SECONDS:-300}" GATEWAY_CONTROL_ADDR="${GATEWAY_CONTROL_ADDR:-127.0.0.1:8090}" GATEWAY_CMPP_ADDR="${GATEWAY_CMPP_ADDR:-0.0.0.0:17890}" CMPP_PUBLIC_HOST="${CMPP_PUBLIC_HOST:-8.160.169.106}" @@ -182,7 +189,7 @@ SQL write_env() { log "Writing production environment" - mkdir -p /etc/cmpp-platform "$APP_DIR" "$APP_DIR/logs/api" "$APP_DIR/logs/gateway" "$APP_DIR/backups" "$OBJECT_STORAGE_LOCAL_ROOT" + mkdir -p /etc/cmpp-platform "$APP_DIR" "$APP_DIR/logs/api" "$APP_DIR/logs/send-worker" "$APP_DIR/logs/gateway" "$APP_DIR/backups" "$OBJECT_STORAGE_LOCAL_ROOT" cat >/etc/cmpp-platform/cmpp-platform.env </etc/systemd/system/cmpp-send-worker.service <&2 + exit 1 +fi + +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 [[ -z "${CMPP_PUBLIC_HOST:-}" || ! "${CMPP_PUBLIC_PORT:-}" =~ ^[1-9][0-9]*$ ]]; then echo "CMPP_PUBLIC_HOST and a positive CMPP_PUBLIC_PORT are required in $ENV_FILE; these are the customer-facing CMPP endpoint." >&2 exit 1 @@ -64,7 +74,35 @@ PROD_ADMIN_CREDENTIAL_FILE="$ADMIN_CREDENTIAL_FILE" node tools/deploy/ensure-pro chmod 600 "$ADMIN_CREDENTIAL_FILE" || true echo "[deploy] Ensuring runtime log directories" -install -d -m 0755 "$APP_DIR/logs/api" "$APP_DIR/logs/gateway" +install -d -m 0755 "$APP_DIR/logs/api" "$APP_DIR/logs/send-worker" "$APP_DIR/logs/gateway" + +echo "[deploy] Installing split API and send-worker services" +node_bin="$(command -v node)" +install -d -m 0755 /etc/systemd/system/cmpp-api.service.d +cat >/etc/systemd/system/cmpp-api.service.d/process-role.conf <<'EOF' +[Service] +Environment=CMPP_PROCESS_ROLE=api +EOF +cat >/etc/systemd/system/cmpp-send-worker.service </dev/null || true - systemctl enable --now cmpp-api cmpp-gateway nginx + systemctl enable --now cmpp-api cmpp-send-worker cmpp-gateway nginx else systemctl enable --now cmpp-minio systemctl restart cmpp-minio - systemctl enable --now cmpp-api cmpp-gateway nginx + systemctl enable --now cmpp-api cmpp-send-worker cmpp-gateway nginx fi systemctl restart cmpp-gateway systemctl restart cmpp-security-agent systemctl restart cmpp-api +systemctl restart cmpp-send-worker systemctl restart nginx echo "[deploy] Health checks" @@ -117,6 +156,7 @@ wait_for_http() { return 1 } wait_for_http "API" "http://127.0.0.1:${API_PORT:-3000}/api/health" +wait_for_http "Send worker metrics" "http://127.0.0.1:${API_WORKER_METRICS_PORT:-9465}/metrics" wait_for_http "Gateway" "http://127.0.0.1:8090/health" redis-cli -h "${REDIS_HOST:-127.0.0.1}" -p "${REDIS_PORT:-6379}" ping >/dev/null pg_isready -d "${DATABASE_URL%%\?*}" >/dev/null diff --git a/tools/deploy/verify-production-deployment.mjs b/tools/deploy/verify-production-deployment.mjs index 9dd0d1a..3349301 100644 --- a/tools/deploy/verify-production-deployment.mjs +++ b/tools/deploy/verify-production-deployment.mjs @@ -4,6 +4,7 @@ import { resolve } from 'node:path'; const deploy = readFileSync(resolve(import.meta.dirname, 'production-deploy.sh'), 'utf8'); const bootstrap = readFileSync(resolve(import.meta.dirname, 'production-bootstrap.sh'), 'utf8'); const apiMain = readFileSync(resolve(import.meta.dirname, '../../api/src/main.ts'), 'utf8'); +const workerMain = readFileSync(resolve(import.meta.dirname, '../../api/src/send-worker.ts'), 'utf8'); const required = [ 'compression_config=/etc/nginx/conf.d/cmpp-compression.conf', ': >"$compression_config"', @@ -26,4 +27,22 @@ if (!apiMain.includes("process.env.API_HOST?.trim() || '127.0.0.1'") || !apiMain throw new Error('NestJS API must bind to API_HOST and default to loopback'); } -console.log('Production deployment verified: Nginx compression is idempotent and NestJS defaults to loopback.'); +for (const marker of [ + 'CMPP_INBOUND_FAST_PATH_ENABLED=true', + 'CMPP_INBOUND_WORKFLOW_WORKER_ENABLED=true', + 'API_INBOUND_WORKFLOW_CONCURRENCY', + 'cmpp-send-worker.service', + 'Environment=CMPP_PROCESS_ROLE=api', + 'Environment=CMPP_PROCESS_ROLE=worker', + 'dist/send-worker.js', + 'API_WORKER_METRICS_PORT', +]) { + if (!`${deploy}\n${bootstrap}`.includes(marker)) { + throw new Error(`production deployment is missing the durable inbound worker contract: ${marker}`); + } +} +if (!workerMain.includes("process.env.API_WORKER_METRICS_HOST?.trim() || '127.0.0.1'")) { + throw new Error('send worker metrics must default to loopback'); +} + +console.log('Production deployment verified: Nginx/API guards and the split durable send worker contract are present.'); diff --git a/tools/monitoring/cmpp-alerts.yml b/tools/monitoring/cmpp-alerts.yml index a5812e2..f3370f6 100644 --- a/tools/monitoring/cmpp-alerts.yml +++ b/tools/monitoring/cmpp-alerts.yml @@ -182,7 +182,7 @@ groups: - name: cmpp-core-services rules: - alert: CmppCoreServiceInactive - expr: node_systemd_unit_state{name=~"cmpp-api\\.service|cmpp-gateway\\.service|postgresql\\.service|redis(-server)?\\.service|cmpp-minio\\.service|nginx\\.service",state="active"} == 0 + expr: node_systemd_unit_state{name=~"cmpp-api\\.service|cmpp-send-worker\\.service|cmpp-gateway\\.service|postgresql\\.service|redis(-server)?\\.service|cmpp-minio\\.service|nginx\\.service",state="active"} == 0 for: 2m labels: severity: critical @@ -200,6 +200,16 @@ groups: for: 2m labels: { severity: critical, service: api } annotations: { summary: "API指标采集不可用", description: "Prometheus连续2分钟无法读取API内部指标端点。", currentValue: "{{ $value }}", threshold: "up = 1" } + - alert: CmppSendWorkerMetricsDown + expr: up{job="cmpp-send-worker"} == 0 + for: 2m + labels: { severity: critical, service: send-worker } + annotations: { summary: "发送Worker指标采集不可用", description: "Prometheus连续2分钟无法读取独立发送Worker指标端点。", currentValue: "{{ $value }}", threshold: "up = 1" } + - alert: CmppInboundWorkflowBacklogCritical + expr: cmpp_worker_inbound_workflow_oldest_pending_age_seconds > 120 + for: 2m + labels: { severity: critical, service: send-worker } + annotations: { summary: "CMPP耐久Inbox严重积压", description: "最旧待处理CMPP Inbox连续2分钟超过120秒。", currentValue: "{{ printf \"%.0f\" $value }}s", threshold: "120s" } - alert: CmppApiHttpErrorRateWarning expr: (sum(rate(cmpp_api_http_requests_total{status=~"5.."}[5m])) / clamp_min(sum(rate(cmpp_api_http_requests_total[5m])), 0.001) > 0.01) and (sum(rate(cmpp_api_http_requests_total{status=~"5.."}[5m])) / clamp_min(sum(rate(cmpp_api_http_requests_total[5m])), 0.001) <= 0.05) and sum(increase(cmpp_api_http_requests_total{status=~"5.."}[5m])) >= 5 for: 5m diff --git a/tools/monitoring/install-prometheus-monitoring.sh b/tools/monitoring/install-prometheus-monitoring.sh index c937ef5..d0a20bd 100644 --- a/tools/monitoring/install-prometheus-monitoring.sh +++ b/tools/monitoring/install-prometheus-monitoring.sh @@ -68,7 +68,7 @@ EOF cat >/etc/systemd/system/prometheus-node-exporter.service.d/cmpp-monitoring.conf </etc/fail2ban/action.d/cmpp-report-only.conf @@ -29,6 +30,18 @@ table inet cmpp_security { chain input { type filter hook input priority -10; policy accept; ip saddr @blocked_ipv4 drop; ip6 saddr @blocked_ipv6 drop; } } EOF + +install -d -m 0755 /etc/systemd/system/cmpp-send-worker.service.d +cat >/etc/systemd/system/cmpp-send-worker.service.d/security-boundary.conf <>/etc/nftables.conf nft -c -f /etc/nftables.conf nft list table inet cmpp_security >/dev/null 2>&1 || nft -f /etc/nftables.d/cmpp-security.nft @@ -54,4 +67,4 @@ fail2ban-client -t nginx -t systemctl daemon-reload systemctl enable cmpp-security-agent -echo "Security boundary installed. Restart cmpp-security-agent and cmpp-api only in the approved release window." +echo "Security boundary installed. Restart cmpp-security-agent, cmpp-api and cmpp-send-worker only in the approved release window."