diff --git a/.env.example b/.env.example index 31e7610..969a36b 100644 --- a/.env.example +++ b/.env.example @@ -42,4 +42,5 @@ GATEWAY_CMPP_VERSION=3.0 GATEWAY_CMPP_ADDR=127.0.0.1:7890 GATEWAY_CMPP_USER=900001 GATEWAY_CMPP_PASSWORD=888888 +GATEWAY_SUBMIT_WORKER_CONCURRENCY=64 CMPP_DOWNSTREAM_ACK_TIMEOUT_SECONDS=30 diff --git a/api/src/metrics/metrics.service.spec.ts b/api/src/metrics/metrics.service.spec.ts index dcc9527..5b42e48 100644 --- a/api/src/metrics/metrics.service.spec.ts +++ b/api/src/metrics/metrics.service.spec.ts @@ -5,11 +5,14 @@ describe('MetricsService', () => { const service = new MetricsService(); const startedAt = service.beginRequest(); service.finishRequest(startedAt, 'GET', '/api/admin/tenants/:id', 200); + const inboundStartedAt = service.beginCmppInboundStage(); + service.finishCmppInboundStage(inboundStartedAt, 'application_lookup', 'success'); const output = service.render(); expect(output).toContain('cmpp_api_process_resident_memory_bytes'); expect(output).toContain('cmpp_api_http_requests_total{method="GET",route="/api/admin/tenants/:id",status="200"} 1'); expect(output).toContain('cmpp_api_http_request_duration_seconds_bucket'); + expect(output).toContain('cmpp_api_cmpp_inbound_stage_duration_seconds_count{stage="application_lookup",result="success"} 1'); expect(output).not.toContain('phone_number'); service.onModuleDestroy(); }); diff --git a/api/src/metrics/metrics.service.ts b/api/src/metrics/metrics.service.ts index 10fc982..231365a 100644 --- a/api/src/metrics/metrics.service.ts +++ b/api/src/metrics/metrics.service.ts @@ -2,6 +2,24 @@ import { Injectable, OnModuleDestroy } from '@nestjs/common'; import { monitorEventLoopDelay } from 'node:perf_hooks'; const HTTP_DURATION_BUCKETS = [0.05, 0.1, 0.25, 0.5, 1, 3, 10] as const; +const CMPP_INBOUND_DURATION_BUCKETS = [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 3, 10] as const; + +export type CmppInboundStage = + | 'application_lookup' + | 'long_message_fragment' + | 'submission_precheck' + | 'template_match' + | 'task_persist' + | 'api_request_persist' + | 'content_detection' + | 'message_persist' + | 'risk_frequency' + | 'billing' + | 'queue_publish' + | 'complete_submit' + | 'total'; + +export type CmppInboundStageResult = 'success' | 'error'; type HttpMetric = { count: number; @@ -25,6 +43,7 @@ export class MetricsService implements OnModuleDestroy { private readonly startedAt = process.hrtime.bigint(); private readonly eventLoopDelay = monitorEventLoopDelay({ resolution: 20 }); private readonly http = new Map(); + private readonly cmppInbound = new Map(); private inFlight = 0; constructor() { @@ -52,6 +71,26 @@ export class MetricsService implements OnModuleDestroy { this.http.set(key, metric); } + beginCmppInboundStage() { + return process.hrtime.bigint(); + } + + finishCmppInboundStage(startedAt: bigint, stage: CmppInboundStage, result: CmppInboundStageResult) { + const key = `${stage}\u0000${result}`; + const metric = this.cmppInbound.get(key) ?? { + count: 0, + durationSum: 0, + buckets: CMPP_INBOUND_DURATION_BUCKETS.map(() => 0), + }; + const durationSeconds = Number(process.hrtime.bigint() - startedAt) / 1_000_000_000; + metric.count += 1; + metric.durationSum += durationSeconds; + CMPP_INBOUND_DURATION_BUCKETS.forEach((bucket, index) => { + if (durationSeconds <= bucket) metric.buckets[index] += 1; + }); + this.cmppInbound.set(key, metric); + } + render() { const memory = process.memoryUsage(); const uptime = Number(process.hrtime.bigint() - this.startedAt) / 1_000_000_000; @@ -78,6 +117,8 @@ export class MetricsService implements OnModuleDestroy { '# TYPE cmpp_api_http_requests_total counter', '# HELP cmpp_api_http_request_duration_seconds API request duration.', '# 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', ]; for (const [key, metric] of this.http) { const [method, route, status] = key.split('\u0000'); @@ -90,6 +131,16 @@ export class MetricsService implements OnModuleDestroy { lines.push(metricLine('cmpp_api_http_request_duration_seconds_sum', metric.durationSum, labels)); lines.push(metricLine('cmpp_api_http_request_duration_seconds_count', metric.count, labels)); } + for (const [key, metric] of this.cmppInbound) { + const [stage, result] = key.split('\u0000'); + const labels = { stage, result }; + CMPP_INBOUND_DURATION_BUCKETS.forEach((bucket, index) => { + lines.push(metricLine('cmpp_api_cmpp_inbound_stage_duration_seconds_bucket', metric.buckets[index], { ...labels, le: String(bucket) })); + }); + lines.push(metricLine('cmpp_api_cmpp_inbound_stage_duration_seconds_bucket', metric.count, { ...labels, le: '+Inf' })); + 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)); + } this.eventLoopDelay.reset(); return `${lines.join('\n')}\n`; } diff --git a/api/src/send-chain/gateway-events.controller.spec.ts b/api/src/send-chain/gateway-events.controller.spec.ts index eaa2a8b..06e794d 100644 --- a/api/src/send-chain/gateway-events.controller.spec.ts +++ b/api/src/send-chain/gateway-events.controller.spec.ts @@ -11,7 +11,17 @@ describe('GatewayEventsController protocol logging', () => { const protocolLogs = { record: jest.fn(), }; - const controller = new GatewayEventsController(sendChain as never, {} as never, protocolLogs as never, { recordEvent: jest.fn() } as never); + const metrics = { + beginCmppInboundStage: jest.fn().mockReturnValue(1n), + finishCmppInboundStage: jest.fn(), + }; + const controller = new GatewayEventsController( + sendChain as never, + {} as never, + protocolLogs as never, + { recordEvent: jest.fn() } as never, + metrics as never, + ); beforeEach(() => { jest.clearAllMocks(); @@ -77,6 +87,20 @@ describe('GatewayEventsController protocol logging', () => { })); }); + it('records a failed inbound total without swallowing the service error', async () => { + const failure = new Error('inbound failed'); + (sendChain as Record).submitInboundMessage = jest.fn().mockRejectedValue(failure); + + await expect(controller.submitInbound({ + account: '607532', + phoneNumber: '13127620092', + content: 'failed', + sequenceId: 142, + })).rejects.toBe(failure); + + expect(metrics.finishCmppInboundStage).toHaveBeenCalledWith(1n, 'total', 'error'); + }); + it('accepts only safe outbound Gateway packet events', () => { expect(controller.protocolLog({ protocol: 'cmpp', @@ -155,6 +179,7 @@ describe('GatewayEventsController protocol logging', () => { messageId: 'MSG-1', status: 'success', })); + expect(metrics.finishCmppInboundStage).toHaveBeenCalledWith(1n, 'total', 'success'); }); it('replaces a fallback receipt identifier with the resolved main message identifier', async () => { diff --git a/api/src/send-chain/gateway-events.controller.ts b/api/src/send-chain/gateway-events.controller.ts index 117c52f..db90766 100644 --- a/api/src/send-chain/gateway-events.controller.ts +++ b/api/src/send-chain/gateway-events.controller.ts @@ -1,4 +1,4 @@ -import { BadRequestException, Body, Controller, Post } from '@nestjs/common'; +import { BadRequestException, Body, Controller, Optional, Post } from '@nestjs/common'; import { ApiTags } from '@nestjs/swagger'; import { GatewayInboundAuthDto, @@ -19,6 +19,7 @@ import { GatewayDownstreamConnectionEventDto } from '../sms-config/sms-config.co import { SmsConfigService } from '../sms-config/sms-config.service'; import { ProtocolLogsService, type ProtocolLogInput } from '../protocol-logs/protocol-logs.service'; import { SecurityDetectionService } from '../security-detection/security-detection.service'; +import { MetricsService } from '../metrics/metrics.service'; @ApiTags('gateway-events') @Controller('gateway/events') @@ -28,6 +29,7 @@ export class GatewayEventsController { private readonly smsConfig: SmsConfigService, private readonly protocolLogs: ProtocolLogsService, private readonly security: SecurityDetectionService, + @Optional() private readonly metrics?: MetricsService, ) {} @Post('submit-result') @@ -139,6 +141,9 @@ export class GatewayEventsController { direction: ProtocolLogInput['direction'] = 'channel_to_platform', ) { const startedAt = Date.now(); + const inboundMetricStartedAt = eventType === 'submit' && direction === 'client_to_platform' + ? this.metrics?.beginCmppInboundStage() + : undefined; const value = body as Record; const common: Omit = { protocol: 'cmpp', @@ -173,6 +178,9 @@ export class GatewayEventsController { status: 'success', durationMs: Date.now() - startedAt, }); + if (inboundMetricStartedAt != null) { + this.metrics?.finishCmppInboundStage(inboundMetricStartedAt, 'total', 'success'); + } return result; } catch (error) { this.protocolLogs.record({ @@ -181,6 +189,9 @@ export class GatewayEventsController { durationMs: Date.now() - startedAt, detail: { error: error instanceof Error ? error.message : 'unknown error' }, }); + if (inboundMetricStartedAt != null) { + this.metrics?.finishCmppInboundStage(inboundMetricStartedAt, 'total', 'error'); + } throw error; } } diff --git a/api/src/send-chain/send-chain.service.ts b/api/src/send-chain/send-chain.service.ts index 0cda40c..05b8d52 100644 --- a/api/src/send-chain/send-chain.service.ts +++ b/api/src/send-chain/send-chain.service.ts @@ -12,6 +12,7 @@ import { PhoneRoutingLookupService } from '../dictionaries/phone-routing-lookup. import { PrismaService } from '../prisma/prisma.service'; import { RiskReviewService } from '../risk-review/risk-review.service'; import { PhoneFrequencyService } from '../risk-review/phone-frequency.service'; +import { MetricsService } from '../metrics/metrics.service'; import { OpenApiService } from '../open-api/open-api.service'; import type { CreateBatchTaskDto, CreateHttpBatchTaskDto, GatewayInboundAuthDto, GatewayInboundSubmitDto, GatewayInboundSingleSubmitResult, GatewaySubmitResultDto, GatewaySubmitSegmentResultDto, GatewayReceiptEventDto, GatewayUplinkEventDto, UplinkMatchCandidateInput, GatewayPendingDeliveryQueryDto, GatewayDownstreamSentDto, GatewayDownstreamAcknowledgedDto, GatewayDownstreamFailureType, GatewayControlDeliveryResult, GatewaySubmitDeadLetterDto, RequeueGatewaySubmitExceptionDto, GatewayDownstreamRecoveryStatusDto, TimeoutUnknownDto, ImportPreviewDto, ConfirmImportDto, SendJob, QueuePriority, RoutedChannel } from './send-chain.contracts'; import { SEND_QUEUE, GATEWAY_SUBMIT_QUEUE, GATEWAY_SUBMIT_STREAM, DEFAULT_DOWNSTREAM_RETRY_DELAY_MS, DEFAULT_DOWNSTREAM_RETRY_MAX_DELAY_MS, DEFAULT_DOWNSTREAM_MAX_RETRIES, DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS, DEFAULT_RECEIPT_TIMEOUT_HOURS, DEFAULT_RECEIPT_TIMEOUT_SCAN_INTERVAL_MS, RECEIPT_TIMEOUT_INITIAL_DELAY_MS, DEFAULT_SCHEDULED_DISPATCH_SCAN_INTERVAL_MS, DEFAULT_SCHEDULED_DISPATCH_STALE_MS, SCHEDULED_DISPATCH_INITIAL_DELAY_MS, DEFAULT_GATEWAY_SUBMIT_REQUEUE_STALE_MS, DEFAULT_DOWNSTREAM_MANUAL_REQUEUE_STALE_MS, DEFAULT_INBOUND_LONG_MESSAGE_SCAN_INTERVAL_MS, INBOUND_LONG_MESSAGE_SCAN_INITIAL_DELAY_MS, DEFAULT_INBOUND_LONG_MESSAGE_PROCESSING_STALE_SECONDS, DEFAULT_UPSTREAM_RECEIPT_INBOX_SCAN_INTERVAL_MS, UPSTREAM_RECEIPT_INBOX_INITIAL_DELAY_MS, DEFAULT_UPSTREAM_RECEIPT_INBOX_PROCESSING_STALE_MS, DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_ATTEMPTS, DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_AGE_HOURS, GATEWAY_SUBMIT_REQUEUE_IDEMPOTENCY_TTL_SECONDS, BULLMQ_PRIORITY, gatewaySubmitRequeueKey, drainageRejectionReason, statusFromRisk, parseSchedule, isObjectRecord, asDateOrNull, downstreamRetryDelayMs, downstreamAckTimeoutMs, downstreamRetryBaseDelayMs, downstreamRetryMaxDelayMs, downstreamMaxRetries, downstreamPendingTimeoutHours, downstreamControlFailureMessage, parseImportRows, splitImportLine, cellByHeader, normalizeCarrier, normalizeQueuePriority, getPositiveConfigInteger, getNonNegativeConfigInteger, isCarrierCompatible, normalizeRegion, matchTemplateContent, escapeRegularExpression, isNationalChannel, isProvinceChannel, validateInboundApplicationSrcId, composeUpstreamSrcId, positiveInteger, parseOptionalSequenceId, normalizeSubmitStatus, normalizeReceiptStatus, downstreamDeliveryAttemptKey, shanghaiDateKey, bullmqConnection, matchesApplicationSecret, octetString, hasRecoveryAuditStateChanged, normalizeRecoveryFailureCategory } from './send-chain.helpers'; @@ -50,6 +51,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { private readonly phoneFrequency: PhoneFrequencyService, @Optional() @Inject(forwardRef(() => OpenApiService)) private readonly openApi?: OpenApiService, @Optional() phoneRouting?: PhoneRoutingLookupService, + @Optional() metrics?: MetricsService, ) { const resolvedPhoneRouting = phoneRouting ?? new PhoneRoutingLookupService(prisma); this.submission = new SendSubmissionService( @@ -64,6 +66,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy { recordCmppFailureReceipt: (message, errorCode, reason) => this.recordCmppFailureReceipt(message, errorCode, reason), }, + metrics, ); this.completion = new SendCompletionService( prisma, diff --git a/api/src/send-chain/send-inbound-entry.service.ts b/api/src/send-chain/send-inbound-entry.service.ts index c8d353e..76036f5 100644 --- a/api/src/send-chain/send-inbound-entry.service.ts +++ b/api/src/send-chain/send-inbound-entry.service.ts @@ -11,6 +11,7 @@ import { PhoneRoutingLookupService } from '../dictionaries/phone-routing-lookup. import { PrismaService } from '../prisma/prisma.service'; import { RiskReviewService } from '../risk-review/risk-review.service'; import { PhoneFrequencyService } from '../risk-review/phone-frequency.service'; +import { MetricsService, type CmppInboundStage } from '../metrics/metrics.service'; import type { CreateBatchTaskDto, CreateHttpBatchTaskDto, GatewayInboundAuthDto, GatewayInboundSubmitDto, GatewayInboundSingleSubmitResult, ImportPreviewDto, ConfirmImportDto, SendJob, QueuePriority, RoutedChannel } from './send-chain.contracts'; import { SEND_QUEUE, GATEWAY_SUBMIT_QUEUE, GATEWAY_SUBMIT_STREAM, DEFAULT_SCHEDULED_DISPATCH_STALE_MS, DEFAULT_INBOUND_LONG_MESSAGE_PROCESSING_STALE_SECONDS, GATEWAY_SUBMIT_REQUEUE_IDEMPOTENCY_TTL_SECONDS, BULLMQ_PRIORITY, statusFromRisk, parseSchedule, parseImportRows, splitImportLine, cellByHeader, normalizeCarrier, normalizeQueuePriority, getPositiveConfigInteger, getNonNegativeConfigInteger, isCarrierCompatible, matchTemplateContent, isNationalChannel, validateInboundApplicationSrcId, composeUpstreamSrcId, positiveInteger, parseOptionalSequenceId, shanghaiDateKey, bullmqConnection, matchesApplicationSecret, octetString, selectChannelCandidate } from './send-chain.helpers'; import type { SendSubmissionCallbacks, SendSubmissionService } from './send-submission.service'; @@ -30,8 +31,21 @@ export class SendInboundEntryService { private readonly phoneRouting: PhoneRoutingLookupService, private readonly facade: SendSubmissionService, private readonly callbacks: SendSubmissionCallbacks, + private readonly metrics?: MetricsService, ) {} + private async measureInboundStage(stage: CmppInboundStage, action: () => Promise): Promise { + const startedAt = this.metrics?.beginCmppInboundStage(); + try { + const result = await action(); + if (startedAt != null) this.metrics?.finishCmppInboundStage(startedAt, stage, 'success'); + return result; + } catch (error) { + if (startedAt != null) this.metrics?.finishCmppInboundStage(startedAt, stage, 'error'); + throw error; + } + } + private releaseMessageReservation( message: { tenantId: string; batchTaskId: string; messageId: string; amountCents: number | bigint; billingUnits: number }, remark: string, @@ -145,7 +159,10 @@ async submitInboundMessage(data: GatewayInboundSubmitDto) { throw new BadRequestException('CMPP submit phone number is invalid'); } - const application = await this.facade.findInboundApplication(data.account); + const application = await this.measureInboundStage( + 'application_lookup', + () => this.facade.findInboundApplication(data.account), + ); if (!application) { throw new BadRequestException('CMPP account is invalid'); } @@ -157,7 +174,10 @@ async submitInboundMessage(data: GatewayInboundSubmitDto) { throw new BadRequestException('CMPP source IP is not in application allowlist'); } validateInboundApplicationSrcId(data.srcId, application); - const collection = await this.facade.collectInboundLongMessageFragment(data, application, phoneNumbers); + const collection = await this.measureInboundStage( + 'long_message_fragment', + () => this.facade.collectInboundLongMessageFragment(data, application, phoneNumbers), + ); if (collection.response) { return collection.response; } @@ -180,16 +200,18 @@ async submitInboundMessage(data: GatewayInboundSubmitDto) { }; } try { - const response = await this.facade.recoverCompletedInboundLongMessageResponse( - collection.messageId, - phoneNumbers, - ) ?? await this.facade.submitCompleteInboundMessage({ - ...data, - content: collection.content, - sequenceId: collection.sequenceId, - registeredDelivery: collection.registeredDelivery ? 1 : 0, - longMessage: undefined, - }, phoneNumbers, application, collection.messageId); + const response = await this.measureInboundStage('complete_submit', async () => ( + await this.facade.recoverCompletedInboundLongMessageResponse( + collection.messageId, + phoneNumbers, + ) ?? await this.facade.submitCompleteInboundMessage({ + ...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: { @@ -210,7 +232,10 @@ async submitInboundMessage(data: GatewayInboundSubmitDto) { throw error; } } - return this.facade.submitCompleteInboundMessage(data, phoneNumbers, application); + return this.measureInboundStage( + 'complete_submit', + () => this.facade.submitCompleteInboundMessage(data, phoneNumbers, application), + ); } async recoverCompletedInboundLongMessageResponse(messageId: string, phoneNumbers: string[]) { @@ -267,8 +292,9 @@ async submitCompleteInboundMessage( if (!application) { throw new BadRequestException('CMPP account is invalid'); } - const persisted = requestedGroupMessageId - ? await this.prisma.smsMessageRecord.findMany({ + const precheck = await this.measureInboundStage('submission_precheck', async () => { + const persisted = requestedGroupMessageId + ? await this.prisma.smsMessageRecord.findMany({ where: { cmppSubmitGroupMessageId: requestedGroupMessageId, phoneNumber: { in: phoneNumbers }, @@ -284,15 +310,18 @@ async submitCompleteInboundMessage( errorCode: true, }, }) - : []; - const persistedByPhone = new Map(persisted.map((item) => [item.phoneNumber, item])); - const phoneRejections = await this.facade.classifyRejectedPhones(application.tenantId, application.id, phoneNumbers); - const missingPhoneCount = phoneNumbers.filter((phoneNumber) => ( - !persistedByPhone.has(phoneNumber) && !phoneRejections.has(phoneNumber) - )).length; - const dailyQuota = missingPhoneCount > 0 - ? await this.facade.tryReserveDailySendQuota(application.id, missingPhoneCount) - : { reserved: true, dailyLimit: application.dailyLimit ?? 100000 }; + : []; + const persistedByPhone = new Map(persisted.map((item) => [item.phoneNumber, item])); + const phoneRejections = await this.facade.classifyRejectedPhones(application.tenantId, application.id, phoneNumbers); + const missingPhoneCount = phoneNumbers.filter((phoneNumber) => ( + !persistedByPhone.has(phoneNumber) && !phoneRejections.has(phoneNumber) + )).length; + const dailyQuota = missingPhoneCount > 0 + ? await this.facade.tryReserveDailySendQuota(application.id, missingPhoneCount) + : { reserved: true, dailyLimit: application.dailyLimit ?? 100000 }; + return { persistedByPhone, phoneRejections, dailyQuota, missingPhoneCount }; + }); + const { persistedByPhone, phoneRejections, dailyQuota, missingPhoneCount } = precheck; const dailyLimitRejection = dailyQuota.reserved ? undefined : { @@ -523,7 +552,10 @@ async submitInboundSingleMessage( synchronousRejection?: { code: string; reason: string }, receiptRejection?: { code: string; reason: string }, ) { - const application = await this.facade.findInboundApplication(data.account); + const application = await this.measureInboundStage( + 'application_lookup', + () => this.facade.findInboundApplication(data.account), + ); if (!application) { throw new BadRequestException('CMPP account is invalid'); } @@ -531,7 +563,10 @@ async submitInboundSingleMessage( throw new BadRequestException('CMPP source IP is not in application allowlist'); } const clientSrcId = validateInboundApplicationSrcId(data.srcId, application); - const template = await this.facade.resolveInboundTemplateCandidate(application.id, data.content); + const template = await this.measureInboundStage( + 'template_match', + () => this.facade.resolveInboundTemplateCandidate(application.id, data.content), + ); const templateVariables = template ? matchTemplateContent(template.content, data.content) ?? {} : {}; const unitPrice = moneyToNumber(application.customerUnitPrice); const queuePriority = normalizeQueuePriority(application.queuePriority); @@ -542,7 +577,7 @@ async submitInboundSingleMessage( phoneCount: 1, unitPrice, }); - const task = await this.prisma.smsBatchTask.create({ + const task = await this.measureInboundStage('task_persist', () => this.prisma.smsBatchTask.create({ data: { tenantId: application.tenantId, applicationId: application.id, @@ -556,8 +591,8 @@ async submitInboundSingleMessage( rejectReason: synchronousRejection?.reason, progressTotal: 1, }, - }); - await this.prisma.smsApiRequest.create({ + })); + await this.measureInboundStage('api_request_persist', () => this.prisma.smsApiRequest.create({ data: { tenantId: application.tenantId, batchTaskId: task.id, @@ -567,9 +602,12 @@ async submitInboundSingleMessage( payloadSummary: { phoneTotal: 1, contentLength: [...data.content].length, account: data.account }, status: synchronousRejection ? 'rejected' : 'accepted', }, - }); - const drainageDetection = await detectDrainageContent(this.prisma, data.content); - const message = await this.prisma.smsMessageRecord.create({ + })); + 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, @@ -592,7 +630,7 @@ async submitInboundSingleMessage( errorCode: synchronousRejection?.code, errorMessage: synchronousRejection?.reason, }, - }); + })); if (synchronousRejection) { return { @@ -614,16 +652,18 @@ async submitInboundSingleMessage( await this.recordCmppFailureReceipt(message, code, reason); }; const queueAfterRiskChecks = async (options: { templateId?: string; signatureId?: string }) => { - const drainage = await this.facade.resolveDrainageInfoMatch(options.signatureId, data.content); - const drainageInfoId = drainage?.id; - const risk = await this.facade.evaluateRiskWithPhoneFrequency({ - tenantId: application.tenantId, - applicationId: application.id, - templateId: options.templateId, - content: data.content, - variables: options.templateId ? templateVariables : undefined, - phoneNumber: data.phoneNumber, - sourceType: 'cmpp', + const { drainageInfoId, risk } = await this.measureInboundStage('risk_frequency', async () => { + const drainage = await this.facade.resolveDrainageInfoMatch(options.signatureId, data.content); + const evaluatedRisk = await this.facade.evaluateRiskWithPhoneFrequency({ + tenantId: application.tenantId, + applicationId: application.id, + templateId: options.templateId, + content: data.content, + variables: options.templateId ? templateVariables : undefined, + phoneNumber: data.phoneNumber, + sourceType: 'cmpp', + }); + return { drainageInfoId: drainage?.id, risk: evaluatedRisk }; }); if (risk.status === 'rejected') { await reject('RISK', risk.reason || '短信被风控拒绝'); @@ -645,32 +685,37 @@ async submitInboundSingleMessage( }); return; } - const accountCheck = await this.billing.checkAccount({ - tenantId: application.tenantId, - amountCents: billing.amountCents, + const accountCheck = await this.measureInboundStage('billing', async () => { + const check = await this.billing.checkAccount({ + tenantId: application.tenantId, + amountCents: billing.amountCents, + }); + if (check.canSend && billing.amountCents > 0) { + await this.billing.freeze({ + tenantId: application.tenantId, + amountCents: billing.amountCents, + relatedType: 'sms_batch_task', + relatedId: task.id, + remark: 'CMPP 入站短信冻结', + }); + } + return check; }); if (!accountCheck.canSend) { await reject('BALANCE', '企业账户余额不足'); return; } - if (billing.amountCents > 0) { - await this.billing.freeze({ - tenantId: application.tenantId, - amountCents: billing.amountCents, - relatedType: 'sms_batch_task', - relatedId: task.id, - remark: 'CMPP 入站短信冻结', + await this.measureInboundStage('queue_publish', async () => { + await this.prisma.smsMessageRecord.update({ + where: { id: message.id }, + data: { status: 'queued', signatureId: options.signatureId, drainageInfoId }, }); - } - await this.prisma.smsMessageRecord.update({ - where: { id: message.id }, - data: { status: 'queued', signatureId: options.signatureId, drainageInfoId }, + await this.prisma.smsBatchTask.update({ + where: { id: task.id }, + data: { status: 'ready', riskTaskId: risk.task?.id, auditStatus: 'approved' }, + }); + await this.facade.enqueueBatchTask(task.id); }); - await this.prisma.smsBatchTask.update({ - where: { id: task.id }, - data: { status: 'ready', riskTaskId: risk.task?.id, auditStatus: 'approved' }, - }); - await this.facade.enqueueBatchTask(task.id); }; if (receiptRejection) { await reject(receiptRejection.code, receiptRejection.reason); diff --git a/api/src/send-chain/send-submission.service.ts b/api/src/send-chain/send-submission.service.ts index 9227552..e5e614e 100644 --- a/api/src/send-chain/send-submission.service.ts +++ b/api/src/send-chain/send-submission.service.ts @@ -4,6 +4,7 @@ import { PhoneRoutingLookupService } from '../dictionaries/phone-routing-lookup. import { PrismaService } from '../prisma/prisma.service'; import { RiskReviewService } from '../risk-review/risk-review.service'; import { PhoneFrequencyService } from '../risk-review/phone-frequency.service'; +import { MetricsService } from '../metrics/metrics.service'; import type { CreateBatchTaskDto, CreateHttpBatchTaskDto, GatewayInboundAuthDto, GatewayInboundSubmitDto, ImportPreviewDto, ConfirmImportDto, SendJob, QueuePriority, RoutedChannel } from './send-chain.contracts'; import { SendBatchEntryService } from './send-batch-entry.service'; import { SendGatewaySubmitService } from './send-gateway-submit.service'; @@ -56,9 +57,10 @@ export class SendSubmissionService { phoneRouting: PhoneRoutingLookupService, facade: SendSubmissionService, callbacks: SendSubmissionCallbacks, + metrics?: MetricsService, ) { this.batchEntry = new SendBatchEntryService(prisma, billing, riskReview, phoneFrequency, phoneRouting, facade, callbacks); - this.inboundEntry = new SendInboundEntryService(prisma, billing, riskReview, phoneFrequency, phoneRouting, facade, callbacks); + this.inboundEntry = new SendInboundEntryService(prisma, billing, riskReview, phoneFrequency, phoneRouting, facade, callbacks, metrics); this.reviewContinuation = new SendReviewContinuationService(prisma, billing, riskReview, phoneFrequency, phoneRouting, facade, callbacks); this.scheduledDispatch = new SendScheduledDispatchService(prisma, billing, riskReview, phoneFrequency, phoneRouting, facade, callbacks); this.gatewaySubmit = new SendGatewaySubmitService(prisma, billing, riskReview, phoneFrequency, phoneRouting, facade, callbacks); diff --git a/docs/codebase-modularization-roadmap.md b/docs/codebase-modularization-roadmap.md index 0cedea5..e9590ce 100644 --- a/docs/codebase-modularization-roadmap.md +++ b/docs/codebase-modularization-roadmap.md @@ -1150,5 +1150,7 @@ global，不能错误归入client。`client-signature-*`、发送页、企业认 - `api/src/metrics/`只负责API进程和HTTP路由模板的低基数计数/直方图，不依赖Prisma、Redis或业务Service；独立回环监听由`api/src/main.ts`组装。 - `gateway/internal/metrics/`只负责线程安全聚合和Prometheus文本暴露；上游连接池、下游Session和Submit Worker仅提供总数或有界结果，不把实体ID或凭据交给监控模块。 +- CMPP性能分段沿用上述边界：业务模块只在原调用边界提交固定阶段名、成功标志和单调时钟耗时；指标模块拒绝未知阶段。`supplier_rtt`在供应商连接调用结束时立即停止，API结果回写另计`api_callback`，防止后续优化依据混杂总耗时。 +- V2提交工作池只归`gateway/internal/submitworker/`治理：Redis领取、全局槽位、在途消息ID和逐条ACK不能渗入上游连接池；`gateway/internal/upstream/`继续只负责通道连接、窗口和供应商协议往返。这样Worker吞吐调优不会改写CMPP连接状态机，连接池也不能自行确认Redis消息。 - `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 8ce1280..19e68b5 100644 --- a/docs/contracts/inbound-r6-declarations.json +++ b/docs/contracts/inbound-r6-declarations.json @@ -78,7 +78,7 @@ "name": "authRequest", "kind": "type", "file": "authentication.go", - "sha256": "0d3e6ab7ec4f418fefb88d2930b01c23eaff926097474c50ba678d553a1aa5a1" + "sha256": "742c3df9a2743741b679a5e1524a8826168385e2af5442fc9019ffc8139ea159" }, { "name": "authResponse", @@ -90,7 +90,7 @@ "name": "authenticate", "kind": "func", "file": "authentication.go", - "sha256": "3dba30340070be6b78dcb1de963fde7a567701925684279c5b675ee9b197ec2c" + "sha256": "fbd2a9aab7116a66f4a9cdc4538d1ccd71f822dda37f3f92c08d62417c1f4418" }, { "name": "cmppVersionName", @@ -102,7 +102,7 @@ "name": "handleLogin", "kind": "func", "file": "authentication.go", - "sha256": "bf3bdbcda7ddfd9f0e2ff53b436e151c94247b4ed40bcf60ec2f78fdb2559e64" + "sha256": "b40ab9a3d1230f71af983a0b1a7c510f56f9e1e1d09817f8705b55c6412ee11f" }, { "name": "setInboundConnectResponse", @@ -348,7 +348,7 @@ "name": "Server", "kind": "type", "file": "server.go", - "sha256": "9dacd52458fb19e19752e201de3b9c9b6e8fd9ba93afda454e7d7444b17148f5" + "sha256": "b43c6128e784432a544631754f449c49fa1ad3e2cd456ccf795cb5fef69fceaf" }, { "name": "defaultHTTPTimeout", @@ -474,7 +474,7 @@ "name": "handleSubmit", "kind": "func", "file": "submit.go", - "sha256": "61e3b8235ab9121a82651e07ce53e38cb650561d80c5f1e4fe6bf4b101b4240a" + "sha256": "76550096e4047d8bcfbb14f693294e86c7a1bf0be2db953f8afebaecd661c577" }, { "name": "inboundLongMessageFragment", @@ -552,7 +552,7 @@ "name": "post", "kind": "func", "file": "transport.go", - "sha256": "10fac868d1fe77fe2c261f14692245b6a08b573ee460cde0c7a77f6ee3ca77b1" + "sha256": "7982e33b65432936ef97e85357e662a1547c4a93a9852f145127b74ab50609c3" }, { "name": "remoteIP", diff --git a/docs/contracts/upstream-r7-declarations.json b/docs/contracts/upstream-r7-declarations.json index 99a2cb8..6c1e917 100644 --- a/docs/contracts/upstream-r7-declarations.json +++ b/docs/contracts/upstream-r7-declarations.json @@ -389,7 +389,7 @@ "kind": "func", "receiver": "Manager", "file": "submit.go", - "sha256": "81d3554f05c38d91cde37b4fa918c2d30d7f4e6da2265c2fe0f9b499650375f3" + "sha256": "a25d30aeb87c1fe123929330fdabc99261e63d5b51e1cb6207197c43a15de07d" }, { "name": "submitPart", @@ -410,7 +410,7 @@ "kind": "func", "receiver": "connectionPool", "file": "submit.go", - "sha256": "9efe1c726215a512d7039c852fdca06544f603ed810daae3fe291cc51038315b" + "sha256": "e3dbf2d433ef4ff95b2fe7a8494b17c6a8e185bdace017cce9c7d2e8c785bfb3" }, { "name": "defaultInt", diff --git a/docs/first-version-development-requirements.md b/docs/first-version-development-requirements.md index 6f9d5c8..9136f0e 100644 --- a/docs/first-version-development-requirements.md +++ b/docs/first-version-development-requirements.md @@ -238,7 +238,7 @@ 7. Gateway 必须实现真实 CMPP Submit，包括短信内容编码、长短信拆分、RegisteredDelivery、serviceId、srcId、destTerminalId、msgFmt、feeType/feeCode 等字段映射。 8. Gateway 必须消费 NestJS 投递的 `SubmitCommand` 队列或等价内部接口；提交成功、提交失败、超时均必须回传 `SubmitResult`，不得只停留在 API 侧入队。 9. Gateway 必须按通道连接和窗口容量控制并发，处理窗口满、SMSC 慢响应、sequence 回绕、连接断开时的在途消息状态。 -10. Gateway 必须对每个物理通道执行 Redis 分布式 TPS 限速。`SmsChannel.rateLimitPerSecond` 是单通道上限，不是平台总上限；同一通道被多个通道组或多个 Gateway 实例使用时共享同一额度，不同通道独立计数。通道连接命令下发的配置值是最终上限，提交命令携带的值只能进一步降低、不能放大该上限。超流速消息必须继续保留在 Redis Stream pending 中等待可用时隙，不能因等待直接标记发送失败；Gateway 重启后仍可由 consumer group 恢复。worker 必须并发处理一个读取批次，让不同通道独立等待，不能因低 TPS 通道造成其他通道队头阻塞；单通道实际并发仍由 Redis 限速和 CMPP 窗口共同约束。 +10. Gateway 必须对每个物理通道执行 Redis 分布式 TPS 限速。`SmsChannel.rateLimitPerSecond` 是单通道上限，不是平台总上限；同一通道被多个通道组或多个 Gateway 实例使用时共享同一额度，不同通道独立计数。通道连接命令下发的配置值是最终上限，提交命令携带的值只能进一步降低、不能放大该上限。超流速消息必须继续保留在 Redis Stream pending 中等待可用时隙，不能因等待直接标记发送失败；Gateway 重启后仍可由 consumer group 恢复。worker 必须使用持续补位的全局有界工作池，任一任务完成后立即领取后续消息，不得以“读取10条、等待整批结束”形成批次屏障；每条消息只在自身提交结果已持久回传或完成死信处理后独立`XACK`。全局并发默认64、可由`GATEWAY_SUBMIT_WORKER_CONCURRENCY`配置且上限1024；单通道实际并发仍由 Redis 限速、连接数和 CMPP 窗口共同约束。恢复pending时必须防止超过`MinIdle`的在途消息被同一进程重复提交。 11. Gateway 不承担业务审核、计费、签名报备、通道组路由、黑名单或敏感词判断；这些由 NestJS 完成，Gateway 只执行已授权通道提交与协议事件回传。 #### 4.8.2 下游客户 CMPP 接入能力 @@ -2096,6 +2096,9 @@ - 运营端展示API请求/错误/延迟/事件循环、Gateway Submit/队列/连接、PostgreSQL连接/死锁、Redis内存/连接/淘汰、Nginx连接/请求和MinIO可用性；指标缺失显示“待采集”，不以0伪装。 - 告警必须使用持续窗口和最低样本量；默认阈值、收敛关系、标签禁止项和性能预算以`docs/prometheus-system-monitoring-design-20260814.md`第9节为准。 - 指标不得包含手机号、短信正文、短信/CMPP/任务ID、密钥、完整URL或SQL原文；不得把时序指标高频写入业务PostgreSQL。 +- CMPP入站性能优化第一步只增加观测，不改变SubmitResp、持久化、风控、计费、路由、入队或回执语义。API必须用固定低基数阶段记录`application_lookup/long_message_fragment/submission_precheck/template_match/task_persist/api_request_persist/content_detection/message_persist/risk_frequency/billing/queue_publish/complete_submit/total`耗时及成功/失败；Gateway入站必须拆分`decode/api_roundtrip/response_write/handler_total`。 +- Gateway供应商下发必须分别记录`stream_wait/rate_limit_wait/connection_wait/supplier_rtt/api_callback`；其中`supplier_rtt`只覆盖供应商连接上的Submit请求与SubmitResp往返，不得包含结果回写API的耗时。阶段名和结果值必须使用代码固定白名单，不得添加手机号、企业、应用、通道、消息、任务或连接ID标签。 +- V2有界工作池必须暴露配置槽位数和当前在途槽位数，使用固定`state=configured|in_flight`标签；该指标用于区分Worker容量耗尽与供应商窗口/限速等待，不得增加通道或消息标签。 - 系统监控标题说明必须明确标注数据来自 Prometheus；“服务关键指标”位于趋势/核心服务区域之后、活动告警之前，并提供统一的“告警阈值设置”入口。安全检测页不重复渲染大号标题，说明文字必须明确标注使用 Fail2ban。 - 告警阈值仅开放固定指标的警告/严重数值，不允许前端提交 PromQL、标签、文件路径或持续时间；必须满足警告值小于严重值。配置以 PostgreSQL 保存版本、期望值、生效值和应用状态，经 `promtool check rules` 校验、同目录原子替换和 Prometheus 热加载成功后才标记生效，失败保留上一生效规则并展示原因。 - 右上角预警中心增加“系统监控告警”，通过独立轻量接口统计 Prometheus 当前 firing/pending 告警及严重数，跳转系统监控活动告警区；任一预警域失败不得清空其他域。 diff --git a/docs/production-deployment.md b/docs/production-deployment.md index 5ec5372..a0d92f8 100644 --- a/docs/production-deployment.md +++ b/docs/production-deployment.md @@ -60,6 +60,7 @@ GATEWAY_CMPP_ADDR=0.0.0.0:17890 CMPP_PUBLIC_HOST=8.160.169.106 CMPP_PUBLIC_PORT=17890 GATEWAY_STARTUP_RECONNECT_DELAY_MS=1000 +GATEWAY_SUBMIT_WORKER_CONCURRENCY=64 OBJECT_STORAGE_DRIVER=minio OBJECT_STORAGE_LOCAL_ROOT=/var/lib/cmpp-platform/object-storage PROD_ADMIN_EMAIL=admin@example.com @@ -79,7 +80,7 @@ PROD_ADMIN_PASSWORD='change-me' HTTP容量边界固定为：NestJS普通JSON/URL-encoded请求体`2 MiB`，仅`/api/client/send/imports/*`使用`25 MiB` JSON解析上限，原始CSV/TSV正文继续由业务层限制为`20 MiB`；Gateway读取NestJS API响应最多`4 MiB`且超限必须明确报错。客户文件导入走`sms.lisglo.com`私有API，因此该虚拟主机的`client_max_body_size`必须不低于`30m`，标准bootstrap配置为`50m`。`api.lisglo.com`只承载单条公网HTTP API、Swagger和健康检查，不承载客户文件导入；不要为导入需求开放私有路由或把NestJS所有JSON接口统一放宽到25MiB。发布前使用`nginx -T`确认最终生效值，不能只检查仓库模板。 -Gateway 的最终 TPS 防线依赖与 API 相同的 Redis。通道连接时会写入 `rate:gateway:channel:config:` 权威上限，实际预约使用 `rate:gateway:channel:`；这些 key 不应在正常发布时清理。多 Gateway 实例必须指向同一 Redis，才能共享单通道额度。超速的 `gateway.submit.commands` 消息会保持在 consumer group pending 中等待，不应通过手工 `XACK` 或删除 Stream 处理积压；先检查通道配置、Redis key、consumer group 和 Gateway 日志。 +Gateway 的最终 TPS 防线依赖与 API 相同的 Redis。通道连接时会写入 `rate:gateway:channel:config:` 权威上限，实际预约使用 `rate:gateway:channel:`；这些 key 不应在正常发布时清理。多 Gateway 实例必须指向同一 Redis，才能共享单通道额度。超速的 `gateway.submit.commands` 消息会保持在 consumer group pending 中等待，不应通过手工 `XACK` 或删除 Stream 处理积压；先检查通道配置、Redis key、consumer group 和 Gateway 日志。V2起Submit Worker使用持续补位有界池并逐条ACK，`GATEWAY_SUBMIT_WORKER_CONCURRENCY`缺省64、最大1024；调整前必须同时核对供应商连接数、窗口、TPS限制、Gateway RSS和`cmpp_gateway_submit_worker_slots`，不能用放大并发绕过通道限速。 服务重启顺序必须是 Gateway 在前、API 在后。API 启动后等待 `GATEWAY_STARTUP_RECONNECT_DELAY_MS`（默认 1 秒），从 PostgreSQL 读取全部 active 通道并重新下发真实连接命令，同时恢复 Gateway 内存连接池和 Redis 权威 TPS key；禁止沿用数据库中重启前的 connected 状态冒充当前连接。 @@ -141,7 +142,7 @@ curl http://127.0.0.1:8090/health 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)=' /etc/cmpp-platform/cmpp-platform.env +grep -E '^(API_ENABLE_SEND_WORKER|API_SEND_WORKER_CONCURRENCY|GATEWAY_SUBMIT_WORKER_CONCURRENCY)=' /etc/cmpp-platform/cmpp-platform.env redis-cli --scan --pattern 'rate:gateway:channel:*' redis-cli XINFO GROUPS gateway.submit.commands ``` diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index b81f02a..bff0cd1 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -4664,6 +4664,18 @@ npm run verify:phase8 | TC-INFRA-MON-026 | 高基数和敏感字段防护 | 检查API/Gateway/Exporter全量metrics文本及Prometheus label names/values | 不存在手机号、短信正文、message/submit/task/channel实体ID、凭据、原始URL或SQL文本 | | TC-INFRA-MON-027 | Recording Rules查询收敛 | 刷新系统监控页并检查Prometheus请求 | 服务卡片只读取`cmpp:service_*`固定聚合，不按卡片开放任意PromQL；缺失指标显示“待采集” | | TC-INFRA-MON-028 | 监控开销对比 | 在同等请求压力下对比开启前后API/Gateway CPU、RSS、P95和吞吐 | 无高基数增长、无业务PostgreSQL高频写入；开销超出预算时暂停发布并调整采集/桶配置 | +| TC-CMPP-PERF-OBS-001 | API入站分段耗时 | 在隔离测试环境提交覆盖成功、同步拒绝、长短信分片和异常的CMPP Submit，抓取API回环metrics | 输出固定`cmpp_api_cmpp_inbound_stage_duration_seconds`直方图；查询、分片、预检、模板、各持久化、检测、风控频次、计费、入队、完整提交及总耗时按实际路径增长，结果仅为`success/error` | +| TC-CMPP-PERF-OBS-002 | Gateway入站分段耗时 | 在隔离测试环境发送合法与非法Submit并抓取Gateway回环metrics | `decode/api_roundtrip/response_write/handler_total`分别增长；失败路径也记录对应阶段，不因指标异常吞掉原协议错误 | +| TC-CMPP-PERF-OBS-003 | Gateway供应商下发分段耗时 | 在隔离测试环境构造Stream等待、限速等待、连接窗口等待、供应商慢响应和API慢回调 | `stream_wait/rate_limit_wait/connection_wait/supplier_rtt/api_callback`可独立区分；`supplier_rtt`在API回调变慢时不等量增长 | +| TC-CMPP-PERF-OBS-004 | 观测标签边界 | 检查API/Gateway新增指标文本和Prometheus时序标签 | 只出现固定`stage/result/le`；不得出现手机号、企业/应用/通道/连接/消息/Submit/任务ID、短信正文或凭据，未知阶段不生成时序 | +| TC-CMPP-PERF-OBS-005 | 纯观测语义回归 | 对比启用埋点前后的同一组CMPP Submit结果、数据库记录、扣费冻结、队列命令及回执 | SubmitResp状态、Msg_Id、多号码独立记录、同步拒绝、异步回执、幂等键和业务调用顺序均不改变；埋点不写PostgreSQL/Redis | +| TC-CMPP-PERF-V2-001 | 持续补位无批次屏障 | 工作池并发设为2，先投递一个阻塞任务和一个快速任务，再投递第三个任务 | 快速任务结束后第三个任务立即开始，不等待第一个慢任务结束；读取批次不形成整批`Wait`屏障 | +| TC-CMPP-PERF-V2-002 | 单消息独立ACK | 同一批次投递一快一慢两条消息，慢任务保持在供应商等待 | 快任务完成后Redis PEL立即只剩慢任务；不得等慢任务结束后整批ACK，也不得在供应商结果回传前提前ACK | +| TC-CMPP-PERF-V2-003 | 全局并发边界 | 分别配置并发1、64、1024和大于1024的值，持续投递超过槽位数的消息 | 同时处理数不超过有效配置；缺省为64，大于1024按1024执行，空闲槽位持续补充 | +| TC-CMPP-PERF-V2-004 | Pending恢复去重 | 让一个供应商调用超过`MinIdle`，同时触发`XAUTOCLAIM`扫描 | 同一进程检测到相同Stream消息ID仍在处理时不重复Submit；原任务结束后按自身结果ACK或进入既有失败/死信流程 | +| TC-CMPP-PERF-V2-005 | Worker槽位指标 | 工作池空闲、部分占用和满载时抓取Gateway metrics | `cmpp_gateway_submit_worker_slots`的`configured/in_flight`与真实配置和在途数一致，不包含通道、消息或客户标识 | +| TC-CMPP-PERF-V2-006 | 失败、死信和重启兼容 | 构造提交失败至最大次数、畸形命令、进程重启后的pending恢复 | 失败次数、死信上报、独立ACK和failure hash清理保持既有语义；重启不丢消息、不把未完成消息误报成功 | +| TC-CMPP-PERF-V2-007 | 隔离环境阶梯持续压测 | 在供应商模拟器、真实API/PostgreSQL/Redis/Gateway链路中，先做100条受控突发，再依次执行10、20、30、40、50条/秒各60秒；每档等待Stream排空并核对数据库、模拟器和Prometheus | 每档SubmitResp拒绝和连接错误为0，Stream最终`pending=0/lag=0`，无死信或服务异常；任一档SubmitResp P95超过5秒、Stream持续增长或服务异常时立即停止升档并保留该档证据，不把未执行档位记为通过 | | TC-GLOBAL-ALERT-001 | 铃铛分域预警菜单 | 准备签名清退未读消息和安全待处置告警后点击右上角铃铛 | 弹层分开显示“签名清退预警”和“安全检测与封禁”，分别展示真实数量和摘要，角标等于两项之和 | | TC-GLOBAL-ALERT-002 | 预警菜单跳转 | 分别点击铃铛中的两个菜单项 | 签名项跳转`/admin/signature-retirement`，安全项跳转`/admin/security-detection`，弹层关闭且对应页面读取真实后端数据 | | TC-GLOBAL-ALERT-003 | 域间故障隔离与轻量轮询 | 分别让一个汇总接口失败并观察30秒轮询请求 | 失败域显示0且另一域数据保留；安全预警使用专用汇总接口，不调用完整overview、规则、代理状态或告警大列表 | diff --git a/docs/testing-progress.md b/docs/testing-progress.md index 3072f9e..ac70659 100644 --- a/docs/testing-progress.md +++ b/docs/testing-progress.md @@ -1,6 +1,6 @@ # 第一版系统化测试进度 -> 环境命名：当前 `8.160.169.106:12026`（Web/API）和 `8.160.169.106:17890`（CMPP 入站）实例统一定义为“预发布环境”。历史记录中涉及该实例的验证、部署和业务页面均按预发布环境理解；`production-deploy.sh`、`NODE_ENV=production` 及正式生产安全/备份规范保留原有技术语义，不代表该实例为正式生产。 +> 环境命名：`8.160.169.106:12026`（Web/API）和 `8.160.169.106:17890`（CMPP 入站）实例统一定义为“预生产环境”；`100.93.204.60`统一定义为“虚拟机测试环境”或“测试机”。“虚拟机”不得再用于指代预生产。历史记录中涉及这两个实例的验证和部署按其明确IP归属理解；`production-deploy.sh`、`NODE_ENV=production`及正式生产安全/备份规范保留原有技术语义，不代表测试机或预生产为正式生产。 ## 2026-08-12 企业签名弹窗、充值回执、通道列表与金额显示优化（已提交、已部署） @@ -3687,3 +3687,39 @@ git diff --check - 监控专项 2 套 8 项、Prisma format/generate、前后端 TypeScript、API 正式编译、Vite 生产构建和 `git diff --check` 通过；Vite 仅保留既有大 chunk 提示。本地真实 PostgreSQL 不可用，`prisma migrate deploy` 返回 Schema engine error，因此没有伪造依赖或把本地 migration 记为通过，migration 已在测试机真实 PostgreSQL 验证。 - 最终浏览器控制会话没有可接管的现有标签页，未绕过图形验证码或另行创建登录会话，因此登录后弹窗滚动和按钮视觉点击未伪报为通过；代码样式契约、真实接口、数据库、Prometheus 与测试机部署均已验收，用户刷新测试机现有登录页面即可查看。 - 本轮没有发送、补发或重投短信，没有创建后台重投任务，没有修改通道账号、密码、启停状态、企业余额或客户连接；`api/tsconfig.build.tsbuildinfo`、`tsconfig.tsbuildinfo`、`outputs/`和空文件 `=` 继续作为受保护项排除提交。 + +# 2026-08-17 CMPP压测优化第一步：纯观测分段（本地未部署） + +- 按首轮压测瓶颈方案先实施V0纯观测，不调整SubmitResp快路径、并发窗口、Stream Worker模型、风控、计费、路由、幂等、事务或供应商回调状态机。API入站增加固定阶段直方图，覆盖应用查询、长短信分片、提交预检、模板匹配、任务/API请求/消息持久化、内容检测、风控与频次、计费、入队、完整提交和总耗时。 +- Gateway入站增加`decode/api_roundtrip/response_write/handler_total`，供应商下发增加`stream_wait/rate_limit_wait/connection_wait/supplier_rtt/api_callback`。供应商RTT只围绕真实连接Submit往返计时，API回写单独计时；阶段和结果均为代码白名单，指标不含手机号、企业、应用、通道、连接、消息、Submit或任务ID。 +- API TypeScript正式构建通过；`metrics.service.spec.ts`与`gateway-events.controller.spec.ts`共8项通过。Gateway全量`go test ./... -count=1`及`go vet ./...`通过，4份Redis Stream消息契约与SendChain R10结构门禁通过；上游R7契约按本轮`Manager.Submit/connectionPool.submit`观测边界同步后通过。 +- 入站R6契约已同步本轮`handleSubmit`实现哈希，但门禁首先被开始前已存在且源码未修改的`authentication.go/authRequest`哈希漂移阻塞；没有为通过本轮门禁而重写或归因该认证声明。`git diff --check`通过，仅输出工作区既有的LF/CRLF提示。 +- SendChain专项大套件执行120项，其中115项通过；5项走真实BullMQ连接时因本机`127.0.0.1:6379`拒绝连接而超过5秒，测试进程同时留下Redis重连句柄。未启动或伪造Redis，因此本轮不把该套件记为全通过；该阻塞不影响两个独立指标专项和TypeScript正式构建证据，后续应在具备真实Redis的隔离测试环境补跑语义回归。 +- 本轮未连接预生产或测试虚拟机、未发起压力流量，未发送、补发或重投短信，未修改通道账号、密码、启停状态、企业余额或客户连接；未提交、未推送、未部署。开始前已有的`api/tsconfig.build.tsbuildinfo`、`tsconfig.tsbuildinfo`、`outputs/`及空文件`=`继续保护，不归因于本轮。 + +# 2026-08-20 CMPP压测优化V2：持续有界Submit Worker（测试环境已发布；预生产保持现状） + +- 根据首轮压测报告中供应商提交峰值约14.24次/秒、Redis Stream lag峰值684、API平均耗时82.5ms且VM资源未饱和的证据，V2只重构Gateway供应商Submit Worker，不提前实施入站SubmitResp快路径或V3/V4异步Outbox。 +- 原实现每次`XREADGROUP Count=10`后启动协程并等待整批全部完成，慢任务形成批次屏障。V2改为默认64槽位、最大1024的持续有界工作池：任一任务结束立即按空闲槽位继续领取；成功、终态拒绝或死信均保持单消息独立ACK，失败消息继续留在PEL按既有`MinIdle/MaxFailures`恢复。 +- Pending恢复对本进程在途消息ID去重，并在`XAUTOCLAIM`返回后回查真实PEL，避免原任务恰好ACK时的竞态重复Submit；该处注释解释了为什么必须同时检查内存活动集合和Redis事实。新增`GATEWAY_SUBMIT_WORKER_CONCURRENCY`及`cmpp_gateway_submit_worker_slots{state=configured|in_flight}`，不增加实体标签。 +- Worker专项连续20轮通过；Gateway全量`go test ./... -count=1`、`go vet ./...`、4份Stream契约、R6/R7与SendChain R10结构门禁通过。R6额外将当前HEAD中未被本轮修改的认证/Server/HTTP声明哈希与契约重新对齐，不归因于V2业务修改。 +- 使用仅监听`127.0.0.1:16379`的本地真实Redis 8.8补跑API观测与SendChain回归，3套121项全部通过；测试后临时Redis已确认停止。Jest仍按仓库既有`--forceExit`口径收尾异步句柄。 +- 2026-08-20发布前核对：本地`HEAD=origin/main=c4f36fc50d7906dfb2f97c881e9ea43c6a64c370`；本段操作目标实际为预生产`8.160.169.106`，其`/opt/cmpp-platform/.deployed-commit=433b2ee56f6016ad8afff1bac73f510b8fd53083`。此前记录中将该目标写成“虚拟机”属于环境称谓错误，现明确更正为“预生产”；API/Gateway/PostgreSQL/Redis/MinIO/Nginx均active。用户要求预生产保持当前状态，不执行回退或后续测试环境发布动作。 +- 预生产发布前恢复资产位于`/opt/cmpp-platform-backups/releases/20260820-100554-before-v2-worker`，包含PostgreSQL、运行源码、环境文件和平台配置；`SHA256SUMS`、数据库gzip及源码/配置tar均验证通过。精确发布包`outputs/cmpp-v2-worker-20260820-101239.tar.gz`共923项、2854923字节，本地及预生产服务器SHA-256均为`cd7bb8d05e7bf9a77ced4522948e0f00b8f25f6b5477eb93037e2a0000b7da5d`，排除了`.env`、`node_modules`、`outputs`、`*.tsbuildinfo`和空文件`=`。 +- 标准全量发布完成前端/API/Gateway构建并将数据库从87条推进到90条migration，但在任何服务重启前被系统安全代理安装闸门阻断：Aliyun Linux 4当前启用仓库没有`fail2ban`包。未擅自增加第三方系统源；随后从恢复资产重新构建并恢复原`433b2ee`的API与前端运行产物，仅重启Gateway启用V2。因此API/前端仍为原运行版本，数据库保留3张向前兼容的监控/安全增量migration，当前运行标识明确写为`433b2ee56f6016ad8afff1bac73f510b8fd53083+gateway-v2.cd7bb8d05e7b`，不伪装为全量工作区已发布。 +- Gateway于北京时间10:27:17重启成功，新PID监听17890，回环健康检查通过；`cmpp_gateway_submit_worker_slots{state="configured"}=64`、`in_flight=0`，`gateway.submit.commands`消费者1、`pending=0`、`lag=0`。原4条真实下游连接受90秒旧心跳租约影响，首次重连被连接上限拒绝并出现2次连接协程`close of closed channel` panic；Gateway进程未退出，旧租约自动过期后4条连接全部恢复，PostgreSQL心跳持续更新。该重启恢复现象作为后续独立稳定性缺陷保留，不通过手工修改客户连接规避。 +- 预生产发布后观察到1条既有真实Stream命令由新Worker接受：总提交耗时68.288ms、Stream等待0.927ms、限速等待0.168ms；供应商RTT样本与API回调样本也已按新指标拆分，处理后PEL和lag均为0。随后30秒被动窗口没有新Stream命令，无法形成V2容量/TPS结论；本轮没有主动发压、发送、补发或重投短信。预生产6项核心服务保持active，Gateway RSS约18MB；完整容量复测必须在`100.93.204.60`虚拟机测试环境使用已确认隔离的供应商模拟器执行。 +- 测试环境发布结果：用户明确指定`100.93.204.60`为V2发布目标，并再次确认该地址才是“虚拟机测试环境”。取得用户提供的`hector`账号授权后完成只读预检：Ubuntu 24.04、原运行标识`482f7ac1ae4c219e47aeaac8735c0584b7d120f2`、90条migration、6/6测试供应商连接、下游连接0、Stream `pending=0/lag=0`。发布前恢复资产位于`/opt/cmpp-platform-backups/releases/20260820-104529-before-v2-worker-test`，包含PostgreSQL、运行源码、环境及平台配置，全部SHA-256、gzip和tar校验通过。 +- 测试机使用同一精确归档`cmpp-v2-worker-20260820-101239.tar.gz`，SHA-256再次核对为`cd7bb8d05e7bf9a77ced4522948e0f00b8f25f6b5477eb93037e2a0000b7da5d`；标准`production-deploy.sh`完成两套依赖安装、安全缓解门禁、Prisma generate/migrate、前端/API/Gateway与安全代理构建、Fail2ban配置测试、Nginx校验、服务重启及健康检查，90条migration无待应用项。测试机运行标识为`c4f36fc50d7906dfb2f97c881e9ea43c6a64c370+workspace.v2.cd7bb8d05e7b`。 +- 发布后API、Gateway、安全代理、PostgreSQL、Redis、MinIO、Nginx、Prometheus及四类Exporter均active；API/Gateway健康、前端HTTP 200，3000/8090/9464继续只监听回环，17890按测试CMPP入口监听。Prometheus真实返回8个target为`up`；Gateway测试供应商连接6/6、下游连接0，`cmpp_gateway_submit_worker_slots{state="configured"}=64`、`in_flight=0`，Stream消费者1、`pending=0`、`lag=0`。发布时间窗API/Gateway/安全代理journal无warning，API/Gateway文件日志无新增ERROR/Exception/panic/fatal。本次只发布和只读验证，没有主动发送、补发或重投短信，也没有修改测试通道账号、密码、启停状态、余额或客户连接。 +- 代码保持未提交、未推送。开始前已有的`api/tsconfig.build.tsbuildinfo`、`tsconfig.tsbuildinfo`、`outputs/`及空文件`=`继续保护；本轮生成的发布包位于既有`outputs/`目录，不扩大提交范围。 + +# 2026-08-20 CMPP压测优化V2：测试环境阶梯复测 + +- 仅对`100.93.204.60`虚拟机测试环境执行隔离压测；供应商端为本地CMPP模拟器`100.91.249.119:17900`，没有连接或改变预生产`8.160.169.106`，没有发送真实短信，也没有修改通道账号、密码、启停状态、企业余额或客户连接。测试环境运行标识复核为`c4f36fc50d7906dfb2f97c881e9ea43c6a64c370+workspace.v2.cd7bb8d05e7b`，API、Gateway、PostgreSQL和Redis最终均为active。 +- 按停止线先执行100条受控突发，再执行10条/秒和20条/秒各60秒。100条突发全部收到成功SubmitResp、无拒绝和连接错误，但P50/P95/P99为2972/5555/5808ms，因此不直接跳到高档；10条/秒实际599条，SubmitResp 599/599成功、无拒绝和连接错误，P50/P95/P99为35/268/1317ms；20条/秒实际1199条，SubmitResp 1199/1199成功、无拒绝和连接错误，P50/P95/P99升至1110/5580/7767ms。20条/秒P95超过5秒停止线，因此没有执行30/40/50条/秒，未把未执行档位记为通过。 +- V2 Worker目标已获得正向证据：10条/秒阶段Stream最大pending 15、lag 0、最大在途15，结束后排空；20条/秒阶段最大pending 43、lag 0、最大在途43，结束后同样`pending=0/lag=0`。20条/秒窗口供应商Submit尝试约1290次，约21.5次/秒，明显高于首轮旧实现约14.24次/秒且没有重现lag 684；供应商阶段平均`stream_wait≈3.55ms`、`rate_limit_wait≈0.15ms`、`connection_wait≈0.006ms`、`supplier_rtt≈149.16ms`。测试窗口主机CPU峰值约53.87%、内存使用峰值约27.49%，不是资源饱和。 +- 新瓶颈位于客户入站SubmitResp路径而不是Redis Stream Worker。1898条业务消息的API分段平均总耗时约256.20ms，其中`risk_frequency≈95.73ms`最大，其后为`queue_publish≈37.69ms`、两次`application_lookup`合计约32.62ms、`template_match≈28.91ms`、`submission_precheck≈17.30ms`；`complete_submit`平均约239.41ms。20条/秒时客户端P95达到5.58秒而API平均仍为0.256秒，结合单连接窗口表明入站连接串行处理/排队仍在放大尾延迟，下一步应实施受窗口约束的连接内并发，或把SubmitResp收敛为“最小校验+幂等持久化”后异步执行风控、计费和路由。 +- 数据库按实际首条`queuedAt=2026-08-20 03:01:47.980`对账，恰好新增1898条：最终delivered 1859、failed 11、submitted 28；28条submitted与模拟器配置的28次不回执一致。供应商模拟器同期收到2048次Submit尝试，其中2013次接受、35次拒绝，发送并收到ACK的回执均为1985次，错误0；Gateway重试解释了尝试数高于业务消息数。客户端进程在收集窗口内看到的receipt数量包含异步到达，不能直接替代数据库和模拟器最终对账。 +- 本轮所有1898条记录仍识别为`mobile`，未覆盖联通、电信，优先级在20条/秒下也未表现出隔离优势：priority P95约5991ms，normal P95约4791ms。因此“修复号段/运营商识别后验证移动、联通、电信六通道容量及优先级隔离”继续保持P1未完成。 +- 本轮新增测试辅助脚本`lg-cmpp-stress-lab/scripts/run-v2-stage.ps1`，只根据档位和持续时间生成一次性客户端配置并调用既有真实CMPP压测客户端，不引入mock、静态结果或localStorage。完整复测报告及原始结果保存在短信平台测试项目；代码仍未提交、未推送，预生产保持原状。 diff --git a/gateway/README.md b/gateway/README.md index 9d5004d..2406fa6 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -11,6 +11,7 @@ - 维护 `messageId -> sequenceId -> gatewayMessageId` 映射。 - 支持断线重连和后续消息继续消费。 - 暴露健康检查和最小指标。 +- Redis Stream Submit Worker 使用持续补位的有界工作池并逐条 ACK；默认并发64，可用`GATEWAY_SUBMIT_WORKER_CONCURRENCY`调整，最大1024。供应商通道的真实上限仍由TPS限速、连接数和CMPP窗口共同决定。 ## 建议骨架 diff --git a/gateway/cmd/gateway/main.go b/gateway/cmd/gateway/main.go index 4f8c3b9..b5ca069 100644 --- a/gateway/cmd/gateway/main.go +++ b/gateway/cmd/gateway/main.go @@ -67,6 +67,7 @@ func main() { worker.Stream = getenv("GATEWAY_SUBMIT_STREAM", "gateway.submit.commands") worker.Group = getenv("GATEWAY_SUBMIT_GROUP", "cmpp-gateway") worker.Consumer = getenv("GATEWAY_SUBMIT_CONSUMER", "gateway-1") + worker.Concurrency = positiveEnvInt("GATEWAY_SUBMIT_WORKER_CONCURRENCY", 64) worker.APIBaseURL = apiBaseURL go func() { log.Printf("cmpp gateway submit worker consuming stream=%s group=%s consumer=%s", worker.Stream, worker.Group, worker.Consumer) @@ -85,6 +86,10 @@ func main() { UpstreamDesired: desired, UpstreamConnected: connected, DownstreamConnected: inbound.ActiveConnectionCount(), SubmitWorkerUp: worker != nil, } + if worker != nil { + snapshot.SubmitWorkerConcurrency = worker.ConfiguredConcurrency() + snapshot.SubmitWorkerInFlight = worker.InFlight() + } if worker == nil || worker.Redis == nil { return snapshot } @@ -141,6 +146,14 @@ func getenv(key string, fallback string) string { return value } +func positiveEnvInt(key string, fallback int) int { + value, err := strconv.Atoi(os.Getenv(key)) + if err != nil || value <= 0 { + return fallback + } + return value +} + func hostname() string { name, err := os.Hostname() if err != nil || name == "" { diff --git a/gateway/internal/inbound/submit.go b/gateway/internal/inbound/submit.go index 838cb42..8f9bef7 100644 --- a/gateway/internal/inbound/submit.go +++ b/gateway/internal/inbound/submit.go @@ -1,6 +1,7 @@ package inbound import ( + "cmpp-platform/gateway/internal/metrics" "context" "crypto/md5" "errors" @@ -56,6 +57,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge if !ok { return true, nil } + handlerStartedAt := time.Now() session := findSessionByConn(packet.Conn) if session == nil || strings.TrimSpace(session.account) == "" { logger.Printf( @@ -63,7 +65,8 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge req.protocol, req.protocol, packet.Conn.Conn.RemoteAddr(), req.sequenceID, "authenticated connection session not found", ) setInboundSubmitResponse(response.Packer, 0, 9) - response.AfterSend = s.submitResponseProtocolLogger("", req.protocol, req.sequenceID, "", "", 0, 9) + responseReadyAt := time.Now() + response.AfterSend = observeInboundSubmitResponse(handlerStartedAt, responseReadyAt, false, s.submitResponseProtocolLogger("", req.protocol, req.sequenceID, "", "", 0, 9)) return false, nil } account := session.account @@ -75,7 +78,8 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge fmt.Sprintf("enterprise code mismatch: expected %s", session.enterpriseCode), ) setInboundSubmitResponse(response.Packer, 0, 9) - response.AfterSend = s.submitResponseProtocolLogger(account, defaultString(session.protocol, req.protocol), req.sequenceID, "", "", 0, 9) + responseReadyAt := time.Now() + response.AfterSend = observeInboundSubmitResponse(handlerStartedAt, responseReadyAt, false, s.submitResponseProtocolLogger(account, defaultString(session.protocol, req.protocol), req.sequenceID, "", "", 0, 9)) return false, nil } phones := make([]string, len(req.destTerminalIDs)) @@ -93,19 +97,23 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge clientProtocol, req.protocol, account, enterpriseCode, remote, req.sequenceID, phone, strings.TrimSpace(req.srcID), req.msgFmt, req.pkNumber, req.pkTotal, len(req.destTerminalIDs), len(req.msgContent), ) + decodeStartedAt := time.Now() content, longMessage, err := decodeInboundSubmitContent(req) + metrics.ObserveInboundStage("decode", err == nil, time.Since(decodeStartedAt)) if err != nil { logger.Printf( "cmpp inbound event=submit_rejected protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s result=9 stage=decode reason=%q", clientProtocol, req.protocol, account, remote, req.sequenceID, phone, err, ) setInboundSubmitResponse(response.Packer, 0, 9) - response.AfterSend = s.submitResponseProtocolLogger(account, clientProtocol, req.sequenceID, phone, "", 0, 9) + responseReadyAt := time.Now() + response.AfterSend = observeInboundSubmitResponse(handlerStartedAt, responseReadyAt, false, s.submitResponseProtocolLogger(account, clientProtocol, req.sequenceID, phone, "", 0, 9)) return false, nil } contentHash := fmt.Sprintf("%x", md5.Sum([]byte(content))) startedAt := time.Now() releaseSubmitBarrier := beginDownstreamSubmitBarrier(packet.Conn) + apiStartedAt := time.Now() result, err := s.submit(remote, submitRequest{ Account: account, PhoneNumber: phone, @@ -118,6 +126,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge RemoteIP: remoteIP(remote), LongMessage: longMessage, }) + metrics.ObserveInboundStage("api_roundtrip", err == nil, time.Since(apiStartedAt)) if err != nil || !result.Accepted { reason := "api returned accepted=false" if err != nil { @@ -133,10 +142,11 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge ) setInboundSubmitResponse(response.Packer, 0, responseResult) protocolLogger := s.submitResponseProtocolLogger(account, clientProtocol, req.sequenceID, phone, result.MessageID, 0, responseResult) - response.AfterSend = func(sendErr error) { + responseReadyAt := time.Now() + response.AfterSend = observeInboundSubmitResponse(handlerStartedAt, responseReadyAt, false, func(sendErr error) { releaseSubmitBarrier() protocolLogger(sendErr) - } + }) return false, nil } gatewayMsgID := messageIDFrom(result.MessageID, req.sequenceID) @@ -175,7 +185,8 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge if current := findSessionByConn(packet.Conn); current != nil && current.report != nil { go current.report(current, "submit", "") } - response.AfterSend = func(sendErr error) { + responseReadyAt := time.Now() + response.AfterSend = observeInboundSubmitResponse(handlerStartedAt, responseReadyAt, true, func(sendErr error) { releaseSubmitBarrier() s.emitProtocolLog(protocolLogEvent{ Protocol: "cmpp", @@ -199,7 +210,7 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge logger.Printf("cmpp inbound event=post_submit_pending_flush_failed account=%s message_id=%s error=%q", account, result.MessageID, err.Error()) } }() - } + }) logger.Printf( "cmpp inbound event=submit_accepted protocol=%s packet_type=%s account=%s remote=%s seq=%d phone=%s dest_count=%d accepted_count=%d result=0 message_id=%s gateway_message_id=%d duration_ms=%d content_chars=%d content_hash=%s", clientProtocol, req.protocol, account, remote, req.sequenceID, phone, len(phones), len(responseMessages), result.MessageID, gatewayMsgID, time.Since(startedAt).Milliseconds(), len([]rune(content)), contentHash, @@ -207,6 +218,16 @@ func (s Server) handleSubmit(response *cmpp.Response, packet *cmpp.Packet, logge return false, nil } +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)) + metrics.ObserveInboundStage("handler_total", accepted && sendErr == nil, time.Since(handlerStartedAt)) + if next != nil { + next(sendErr) + } + } +} + type inboundSubmitPacket struct { protocol string pkTotal uint8 diff --git a/gateway/internal/metrics/metrics.go b/gateway/internal/metrics/metrics.go index 62e758b..dd35b63 100644 --- a/gateway/internal/metrics/metrics.go +++ b/gateway/internal/metrics/metrics.go @@ -14,15 +14,39 @@ var submitAccepted atomic.Uint64 var submitFailed atomic.Uint64 var submitDurationNanoseconds atomic.Uint64 +var durationBuckets = [...]time.Duration{ + 5 * time.Millisecond, + 10 * time.Millisecond, + 25 * time.Millisecond, + 50 * time.Millisecond, + 100 * time.Millisecond, + 250 * time.Millisecond, + 500 * time.Millisecond, + time.Second, + 3 * time.Second, + 10 * time.Second, +} + +type durationHistogram struct { + count atomic.Uint64 + sumNano atomic.Uint64 + buckets [len(durationBuckets)]atomic.Uint64 +} + +var inboundStageHistograms [4][2]durationHistogram +var submitStageHistograms [5][2]durationHistogram + type Snapshot struct { - UpstreamDesired int - UpstreamConnected int - DownstreamConnected int - SubmitWorkerUp bool - QueueAvailable bool - QueuePending int64 - QueueLag int64 - QueueOldestAgeSeconds float64 + UpstreamDesired int + UpstreamConnected int + DownstreamConnected int + SubmitWorkerUp bool + SubmitWorkerConcurrency int + SubmitWorkerInFlight int64 + QueueAvailable bool + QueuePending int64 + QueueLag int64 + QueueOldestAgeSeconds float64 } type SnapshotFunc func(context.Context) Snapshot @@ -36,6 +60,25 @@ func ObserveSubmit(accepted bool, duration time.Duration) { submitDurationNanoseconds.Add(uint64(max(duration, 0))) } +// ObserveInboundStage deliberately accepts only a fixed stage/result vocabulary. +// Entity identifiers would create unbounded Prometheus series during high-volume traffic. +func ObserveInboundStage(stage string, success bool, duration time.Duration) { + stageIndex := inboundStageIndex(stage) + if stageIndex < 0 { + return + } + observeDuration(&inboundStageHistograms[stageIndex][boolIndex(success)], duration) +} + +// ObserveSubmitStage separates queueing, limiting, connection-window, supplier and API callback time. +func ObserveSubmitStage(stage string, success bool, duration time.Duration) { + stageIndex := submitStageIndex(stage) + if stageIndex < 0 { + return + } + observeDuration(&submitStageHistograms[stageIndex][boolIndex(success)], duration) +} + func Handler(load SnapshotFunc) http.Handler { return http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { if request.Method != http.MethodGet || request.URL.Path != "/metrics" { @@ -61,9 +104,18 @@ func Handler(load SnapshotFunc) http.Handler { fmt.Fprintf(response, "# HELP cmpp_gateway_submit_total Upstream submit attempts by bounded result.\n# TYPE cmpp_gateway_submit_total counter\ncmpp_gateway_submit_total{result=\"accepted\"} %d\ncmpp_gateway_submit_total{result=\"failed\"} %d\n", accepted, failed) fmt.Fprintf(response, "# HELP cmpp_gateway_submit_duration_seconds_sum Total upstream submit duration.\n# TYPE cmpp_gateway_submit_duration_seconds_sum counter\ncmpp_gateway_submit_duration_seconds_sum %f\n", float64(submitDurationNanoseconds.Load())/float64(time.Second)) fmt.Fprintf(response, "# HELP cmpp_gateway_submit_duration_seconds_count Total measured upstream submits.\n# TYPE cmpp_gateway_submit_duration_seconds_count counter\ncmpp_gateway_submit_duration_seconds_count %d\n", count) + fmt.Fprint(response, "# HELP cmpp_gateway_inbound_stage_duration_seconds CMPP inbound handler duration by bounded stage and result.\n# TYPE cmpp_gateway_inbound_stage_duration_seconds histogram\n") + for index, stage := range []string{"decode", "api_roundtrip", "response_write", "handler_total"} { + writeDurationHistogram(response, "cmpp_gateway_inbound_stage_duration_seconds", stage, &inboundStageHistograms[index]) + } + fmt.Fprint(response, "# HELP cmpp_gateway_submit_stage_duration_seconds Gateway submit duration by bounded stage and result.\n# TYPE cmpp_gateway_submit_stage_duration_seconds histogram\n") + for index, stage := range []string{"stream_wait", "rate_limit_wait", "connection_wait", "supplier_rtt", "api_callback"} { + writeDurationHistogram(response, "cmpp_gateway_submit_stage_duration_seconds", stage, &submitStageHistograms[index]) + } fmt.Fprintf(response, "# HELP cmpp_gateway_upstream_connections Desired and live supplier connections.\n# TYPE cmpp_gateway_upstream_connections gauge\ncmpp_gateway_upstream_connections{state=\"desired\"} %d\ncmpp_gateway_upstream_connections{state=\"connected\"} %d\n", snapshot.UpstreamDesired, snapshot.UpstreamConnected) fmt.Fprintf(response, "# HELP cmpp_gateway_downstream_connections Authenticated client connections.\n# TYPE cmpp_gateway_downstream_connections gauge\ncmpp_gateway_downstream_connections %d\n", snapshot.DownstreamConnected) fmt.Fprintf(response, "# HELP cmpp_gateway_submit_worker_up Whether the submit worker was initialized.\n# TYPE cmpp_gateway_submit_worker_up gauge\ncmpp_gateway_submit_worker_up %d\n", boolNumber(snapshot.SubmitWorkerUp)) + fmt.Fprintf(response, "# HELP cmpp_gateway_submit_worker_slots Configured and active bounded submit worker slots.\n# TYPE cmpp_gateway_submit_worker_slots gauge\ncmpp_gateway_submit_worker_slots{state=\"configured\"} %d\ncmpp_gateway_submit_worker_slots{state=\"in_flight\"} %d\n", snapshot.SubmitWorkerConcurrency, snapshot.SubmitWorkerInFlight) if snapshot.QueueAvailable { fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_pending Pending entries owned by the consumer group.\n# TYPE cmpp_gateway_submit_queue_pending gauge\ncmpp_gateway_submit_queue_pending %d\n", snapshot.QueuePending) fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_lag Undelivered entries for the consumer group.\n# TYPE cmpp_gateway_submit_queue_lag gauge\ncmpp_gateway_submit_queue_lag %d\n", snapshot.QueueLag) @@ -72,6 +124,76 @@ func Handler(load SnapshotFunc) http.Handler { }) } +func observeDuration(histogram *durationHistogram, duration time.Duration) { + duration = max(duration, 0) + histogram.count.Add(1) + histogram.sumNano.Add(uint64(duration)) + for index, bucket := range durationBuckets { + if duration <= bucket { + histogram.buckets[index].Add(1) + } + } +} + +func writeDurationHistogram(response http.ResponseWriter, metricName string, stage string, histograms *[2]durationHistogram) { + for resultIndex, result := range []string{"failed", "success"} { + histogram := &histograms[resultIndex] + count := histogram.count.Load() + if count == 0 { + continue + } + for index, bucket := range durationBuckets { + fmt.Fprintf(response, "%s_bucket{stage=%q,result=%q,le=%q} %d\n", metricName, stage, result, durationBucketLabel(bucket), histogram.buckets[index].Load()) + } + fmt.Fprintf(response, "%s_bucket{stage=%q,result=%q,le=\"+Inf\"} %d\n", metricName, stage, result, count) + fmt.Fprintf(response, "%s_sum{stage=%q,result=%q} %f\n", metricName, stage, result, float64(histogram.sumNano.Load())/float64(time.Second)) + fmt.Fprintf(response, "%s_count{stage=%q,result=%q} %d\n", metricName, stage, result, count) + } +} + +func durationBucketLabel(bucket time.Duration) string { + return fmt.Sprintf("%g", bucket.Seconds()) +} + +func boolIndex(value bool) int { + if value { + return 1 + } + return 0 +} + +func inboundStageIndex(stage string) int { + switch stage { + case "decode": + return 0 + case "api_roundtrip": + return 1 + case "response_write": + return 2 + case "handler_total": + return 3 + default: + return -1 + } +} + +func submitStageIndex(stage string) int { + switch stage { + case "stream_wait": + return 0 + case "rate_limit_wait": + return 1 + case "connection_wait": + return 2 + case "supplier_rtt": + return 3 + case "api_callback": + return 4 + default: + return -1 + } +} + func boolNumber(value bool) int { if value { return 1 diff --git a/gateway/internal/metrics/metrics_test.go b/gateway/internal/metrics/metrics_test.go index 4436faf..e4ce563 100644 --- a/gateway/internal/metrics/metrics_test.go +++ b/gateway/internal/metrics/metrics_test.go @@ -11,16 +11,28 @@ import ( func TestMetricsHandlerExportsOnlyAggregateGatewayState(t *testing.T) { ObserveSubmit(true, 20*time.Millisecond) + ObserveInboundStage("api_roundtrip", true, 30*time.Millisecond) + ObserveSubmitStage("rate_limit_wait", true, 15*time.Millisecond) request := httptest.NewRequest(http.MethodGet, "/metrics", nil) response := httptest.NewRecorder() Handler(func(context.Context) Snapshot { - return Snapshot{UpstreamDesired: 2, UpstreamConnected: 1, DownstreamConnected: 3, SubmitWorkerUp: true, QueueAvailable: true, QueuePending: 4, QueueLag: 5, QueueOldestAgeSeconds: 12} + return Snapshot{UpstreamDesired: 2, UpstreamConnected: 1, DownstreamConnected: 3, SubmitWorkerUp: true, SubmitWorkerConcurrency: 64, SubmitWorkerInFlight: 7, QueueAvailable: true, QueuePending: 4, QueueLag: 5, QueueOldestAgeSeconds: 12} }).ServeHTTP(response, request) body := response.Body.String() if response.Code != http.StatusOK || !strings.Contains(body, "cmpp_gateway_submit_queue_pending 4") || !strings.Contains(body, "cmpp_gateway_upstream_connections{state=\"connected\"} 1") { t.Fatalf("unexpected metrics response: code=%d body=%s", response.Code, body) } + for _, expected := range []string{ + `cmpp_gateway_inbound_stage_duration_seconds_count{stage="api_roundtrip",result="success"} 1`, + `cmpp_gateway_submit_stage_duration_seconds_count{stage="rate_limit_wait",result="success"} 1`, + `cmpp_gateway_submit_worker_slots{state="configured"} 64`, + `cmpp_gateway_submit_worker_slots{state="in_flight"} 7`, + } { + if !strings.Contains(body, expected) { + t.Fatalf("metrics response is missing %q: %s", expected, body) + } + } for _, forbidden := range []string{"phone_number", "message_id", "channel_id"} { if strings.Contains(body, forbidden) { t.Fatalf("metrics expose forbidden label %q", forbidden) diff --git a/gateway/internal/submitworker/worker.go b/gateway/internal/submitworker/worker.go index 6d58695..f06a1d5 100644 --- a/gateway/internal/submitworker/worker.go +++ b/gateway/internal/submitworker/worker.go @@ -10,6 +10,7 @@ import ( "net/http" "strings" "sync" + "sync/atomic" "time" "cmpp-platform/gateway/internal/metrics" @@ -21,11 +22,12 @@ import ( ) const ( - defaultStream = "gateway.submit.commands" - defaultGroup = "cmpp-gateway" - defaultConsumer = "gateway-1" - defaultMinIdle = 30 * time.Second - defaultMaxFails = 3 + defaultStream = "gateway.submit.commands" + defaultGroup = "cmpp-gateway" + defaultConsumer = "gateway-1" + defaultMinIdle = 30 * time.Second + defaultMaxFails = 3 + defaultConcurrency = 64 ) type Worker struct { @@ -39,11 +41,13 @@ type Worker struct { Consumer string Block time.Duration Count int64 + Concurrency int MinIdle time.Duration MaxFailures int APIBaseURL string HTTPClient *http.Client Logger *log.Logger + inFlight atomic.Int64 } type DeadLetterEvent struct { @@ -78,6 +82,8 @@ func (w *Worker) Run(ctx context.Context) error { if w.Upstream == nil { return fmt.Errorf("upstream manager is required") } + pool := newMessageWorkPool(ctx, w, w.concurrency()) + defer pool.wait() for { if err := w.ensureGroup(ctx); err != nil { if ctx.Err() != nil { @@ -87,7 +93,7 @@ func (w *Worker) Run(ctx context.Context) error { sleep(ctx, 3*time.Second) continue } - if err := w.recoverPending(ctx); err != nil { + if err := w.recoverPending(ctx, pool); err != nil { if ctx.Err() != nil { return ctx.Err() } @@ -95,7 +101,7 @@ func (w *Worker) Run(ctx context.Context) error { sleep(ctx, time.Second) continue } - if err := w.consumeOnce(ctx); err != nil { + if err := w.consumeOnce(ctx, pool); err != nil { if ctx.Err() != nil { return ctx.Err() } @@ -113,12 +119,16 @@ func (w *Worker) ensureGroup(ctx context.Context) error { return err } -func (w *Worker) consumeOnce(ctx context.Context) error { +func (w *Worker) consumeOnce(ctx context.Context, pool *messageWorkPool) error { + available, err := pool.waitForCapacity(ctx) + if err != nil { + return err + } streams, err := w.Redis.XReadGroup(ctx, &redis.XReadGroupArgs{ Group: w.group(), Consumer: w.consumer(), Streams: []string{w.stream(), ">"}, - Count: w.count(), + Count: min(w.count(), int64(available)), Block: w.block(), }).Result() if errors.Is(err, redis.Nil) { @@ -128,23 +138,29 @@ func (w *Worker) consumeOnce(ctx context.Context) error { return err } for _, stream := range streams { - if err := w.processMessages(ctx, stream.Messages); err != nil { - return err + for _, message := range stream.Messages { + if !pool.dispatch(message) { + return fmt.Errorf("gateway submit worker capacity accounting mismatch") + } } } return nil } -func (w *Worker) recoverPending(ctx context.Context) error { +func (w *Worker) recoverPending(ctx context.Context, pool *messageWorkPool) error { start := "0-0" for { + available := pool.available() + if available == 0 { + return nil + } messages, next, err := w.Redis.XAutoClaim(ctx, &redis.XAutoClaimArgs{ Stream: w.stream(), Group: w.group(), Consumer: w.consumer(), MinIdle: w.minIdle(), Start: start, - Count: w.count(), + Count: min(w.count(), int64(available)), }).Result() if errors.Is(err, redis.Nil) { return nil @@ -156,8 +172,13 @@ func (w *Worker) recoverPending(ctx context.Context) error { return nil } w.logf("gateway submit worker reclaimed %d pending message(s)", len(messages)) - if err := w.processMessages(ctx, messages); err != nil { - return err + for _, message := range messages { + // An in-flight command can legitimately exceed MinIdle while waiting on a supplier. + // Rechecking both the local active set and Redis PEL closes the race where the + // original attempt ACKs between XAUTOCLAIM returning and local dispatch. + if err := pool.dispatchRecovered(ctx, message); err != nil { + return err + } } start = next if next == "0-0" { @@ -166,27 +187,111 @@ func (w *Worker) recoverPending(ctx context.Context) error { } } -func (w *Worker) processMessages(ctx context.Context, messages []redis.XMessage) error { - var group sync.WaitGroup - for _, message := range messages { - message := message - group.Add(1) - go func() { - defer group.Done() - if err := w.processMessage(ctx, message); err != nil { - w.logf("gateway submit worker message %s failed: %v", message.ID, err) - } - }() +type messageWorkPool struct { + ctx context.Context + worker *Worker + slots chan struct{} + completed chan struct{} + group sync.WaitGroup + mu sync.Mutex + active map[string]struct{} +} + +func newMessageWorkPool(ctx context.Context, worker *Worker, concurrency int) *messageWorkPool { + return &messageWorkPool{ + ctx: ctx, worker: worker, slots: make(chan struct{}, concurrency), + completed: make(chan struct{}, concurrency), active: make(map[string]struct{}), + } +} + +func (p *messageWorkPool) available() int { + return cap(p.slots) - len(p.slots) +} + +func (p *messageWorkPool) waitForCapacity(ctx context.Context) (int, error) { + for p.available() == 0 { + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-p.completed: + } + } + return p.available(), nil +} + +func (p *messageWorkPool) dispatch(message redis.XMessage) bool { + p.mu.Lock() + if _, exists := p.active[message.ID]; exists { + p.mu.Unlock() + return true + } + select { + case p.slots <- struct{}{}: + p.active[message.ID] = struct{}{} + p.worker.inFlight.Add(1) + p.group.Add(1) + p.mu.Unlock() + case <-p.ctx.Done(): + p.mu.Unlock() + return false + default: + p.mu.Unlock() + return false + } + go func() { + defer func() { + p.mu.Lock() + delete(p.active, message.ID) + p.mu.Unlock() + <-p.slots + p.worker.inFlight.Add(-1) + select { + case p.completed <- struct{}{}: + default: + } + p.group.Done() + }() + if err := p.worker.processMessage(p.ctx, message); err != nil { + p.worker.logf("gateway submit worker message %s failed: %v", message.ID, err) + } + }() + return true +} + +func (p *messageWorkPool) dispatchRecovered(ctx context.Context, message redis.XMessage) error { + p.mu.Lock() + _, active := p.active[message.ID] + p.mu.Unlock() + if active { + return nil + } + pending, err := p.worker.Redis.XPendingExt(ctx, &redis.XPendingExtArgs{ + Stream: p.worker.stream(), Group: p.worker.group(), Start: message.ID, End: message.ID, Count: 1, + }).Result() + if err != nil && !errors.Is(err, redis.Nil) { + return err + } + if len(pending) == 0 { + return nil + } + if !p.dispatch(message) { + return fmt.Errorf("gateway submit worker recovery capacity accounting mismatch") } - group.Wait() return nil } +func (p *messageWorkPool) wait() { + p.group.Wait() +} + func (w *Worker) processMessage(ctx context.Context, message redis.XMessage) error { command, err := CommandFromStreamValues(message.Values) if err != nil { return w.deadLetterMalformedMessage(ctx, message, err) } + if !command.CreatedAt.IsZero() { + metrics.ObserveSubmitStage("stream_wait", true, time.Since(command.CreatedAt)) + } if err := w.handleCommand(ctx, command); err != nil { if ctx.Err() != nil { return ctx.Err() @@ -208,11 +313,14 @@ func (w *Worker) processMessage(ctx context.Context, message redis.XMessage) err func (w *Worker) handleCommand(ctx context.Context, command queue.SubmitCommand) error { startedAt := time.Now() + limitStartedAt := time.Now() if w.Limiter != nil { if _, err := w.Limiter.Wait(ctx, command.ChannelID, command.Route.RateLimitPerSecond); err != nil { + metrics.ObserveSubmitStage("rate_limit_wait", false, time.Since(limitStartedAt)) return err } } + metrics.ObserveSubmitStage("rate_limit_wait", true, time.Since(limitStartedAt)) submit := w.Submit if submit == nil { if w.Upstream == nil { @@ -389,6 +497,21 @@ func (w *Worker) count() int64 { return 10 } +func (w *Worker) concurrency() int { + if w.Concurrency > 0 { + return min(w.Concurrency, 1024) + } + return defaultConcurrency +} + +func (w *Worker) ConfiguredConcurrency() int { + return w.concurrency() +} + +func (w *Worker) InFlight() int64 { + return w.inFlight.Load() +} + func (w *Worker) minIdle() time.Duration { if w.MinIdle > 0 { return w.MinIdle diff --git a/gateway/internal/submitworker/worker_test.go b/gateway/internal/submitworker/worker_test.go index f240be3..44b6fb6 100644 --- a/gateway/internal/submitworker/worker_test.go +++ b/gateway/internal/submitworker/worker_test.go @@ -2,6 +2,7 @@ package submitworker import ( "context" + "sync/atomic" "testing" "time" @@ -121,11 +122,21 @@ func TestMinIdleDefaultsToThirtySeconds(t *testing.T) { } } -func TestProcessMessagesDoesNotLetOneChannelBlockAnother(t *testing.T) { +func TestConcurrencyUsesBoundedDefaultAndMaximum(t *testing.T) { + if got := (&Worker{}).ConfiguredConcurrency(); got != defaultConcurrency { + t.Fatalf("default concurrency = %d, want %d", got, defaultConcurrency) + } + if got := (&Worker{Concurrency: 2048}).ConfiguredConcurrency(); got != 1024 { + t.Fatalf("capped concurrency = %d, want 1024", got) + } +} + +func TestMessageWorkPoolContinuouslyRefillsWithoutWaitingForSlowSibling(t *testing.T) { mr := miniredis.RunT(t) client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) startedA := make(chan struct{}) startedB := make(chan struct{}) + startedC := make(chan struct{}) releaseA := make(chan struct{}) worker := &Worker{ Redis: client, @@ -136,19 +147,17 @@ func TestProcessMessagesDoesNotLetOneChannelBlockAnother(t *testing.T) { <-releaseA case "channel-b": close(startedB) + case "channel-c": + close(startedC) } return queue.SubmitResult{SubmitStatus: "accepted"}, nil }, } - messages := []redis.XMessage{ - {ID: "1-0", Values: submitCommandValues("message-a", "channel-a")}, - {ID: "2-0", Values: submitCommandValues("message-b", "channel-b")}, + pool := newMessageWorkPool(context.Background(), worker, 2) + if !pool.dispatch(redis.XMessage{ID: "1-0", Values: submitCommandValues("message-a", "channel-a")}) || + !pool.dispatch(redis.XMessage{ID: "2-0", Values: submitCommandValues("message-b", "channel-b")}) { + t.Fatal("initial messages were not dispatched") } - done := make(chan struct{}) - go func() { - _ = worker.processMessages(context.Background(), messages) - close(done) - }() select { case <-startedA: case <-time.After(time.Second): @@ -159,11 +168,125 @@ func TestProcessMessagesDoesNotLetOneChannelBlockAnother(t *testing.T) { case <-time.After(200 * time.Millisecond): t.Fatal("channel-b was blocked by channel-a") } - close(releaseA) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if _, err := pool.waitForCapacity(ctx); err != nil { + t.Fatalf("wait for refill capacity: %v", err) + } + if !pool.dispatch(redis.XMessage{ID: "3-0", Values: submitCommandValues("message-c", "channel-c")}) { + t.Fatal("refill message was not dispatched") + } select { - case <-done: - case <-time.After(time.Second): - t.Fatal("message batch did not complete") + case <-startedC: + case <-time.After(200 * time.Millisecond): + t.Fatal("pool waited for the slow sibling instead of refilling its free slot") + } + close(releaseA) + pool.wait() +} + +func TestMessageWorkPoolAcknowledgesFastMessageBeforeSlowSiblingCompletes(t *testing.T) { + mr := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + worker := &Worker{Redis: client, Stream: "gateway.submit.commands", Group: "cmpp-gateway"} + ctx := context.Background() + if err := worker.ensureGroup(ctx); err != nil { + t.Fatalf("ensureGroup: %v", err) + } + for _, entry := range []struct{ id, messageID, channelID string }{ + {"1-0", "message-slow", "channel-slow"}, + {"2-0", "message-fast", "channel-fast"}, + } { + if err := client.XAdd(ctx, &redis.XAddArgs{Stream: worker.stream(), ID: entry.id, Values: submitCommandValues(entry.messageID, entry.channelID)}).Err(); err != nil { + t.Fatalf("xadd %s: %v", entry.id, err) + } + } + streams, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{Group: worker.group(), Consumer: worker.consumer(), Streams: []string{worker.stream(), ">"}, Count: 2}).Result() + if err != nil || len(streams) != 1 || len(streams[0].Messages) != 2 { + t.Fatalf("xreadgroup: streams=%+v err=%v", streams, err) + } + slowStarted := make(chan struct{}) + fastReturned := make(chan struct{}) + releaseSlow := make(chan struct{}) + worker.Submit = func(_ context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) { + if command.ChannelID == "channel-slow" { + close(slowStarted) + <-releaseSlow + } else { + close(fastReturned) + } + return queue.SubmitResult{SubmitStatus: "accepted"}, nil + } + pool := newMessageWorkPool(ctx, worker, 2) + for _, message := range streams[0].Messages { + if !pool.dispatch(message) { + t.Fatalf("message %s was not dispatched", message.ID) + } + } + <-slowStarted + <-fastReturned + deadline := time.Now().Add(time.Second) + for { + pending, pendingErr := client.XPending(ctx, worker.stream(), worker.group()).Result() + if pendingErr != nil { + t.Fatalf("xpending: %v", pendingErr) + } + if pending.Count == 1 { + break + } + if time.Now().After(deadline) { + t.Fatalf("pending count = %d, want 1 while slow sibling is still running", pending.Count) + } + time.Sleep(10 * time.Millisecond) + } + close(releaseSlow) + pool.wait() +} + +func TestPendingRecoveryDoesNotDuplicateAnActiveOrAlreadyAcknowledgedMessage(t *testing.T) { + mr := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + worker := &Worker{Redis: client, Stream: "gateway.submit.commands", Group: "cmpp-gateway", MinIdle: time.Millisecond} + ctx := context.Background() + if err := worker.ensureGroup(ctx); err != nil { + t.Fatalf("ensureGroup: %v", err) + } + if err := client.XAdd(ctx, &redis.XAddArgs{Stream: worker.stream(), ID: "3-0", Values: submitCommandValues("message-active", "channel-active")}).Err(); err != nil { + t.Fatalf("xadd: %v", err) + } + streams, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{Group: worker.group(), Consumer: worker.consumer(), Streams: []string{worker.stream(), ">"}, Count: 1}).Result() + if err != nil || len(streams) != 1 || len(streams[0].Messages) != 1 { + t.Fatalf("xreadgroup: streams=%+v err=%v", streams, err) + } + message := streams[0].Messages[0] + started := make(chan struct{}) + release := make(chan struct{}) + var submits atomic.Int32 + worker.Submit = func(_ context.Context, _ queue.SubmitCommand) (queue.SubmitResult, error) { + submits.Add(1) + close(started) + <-release + return queue.SubmitResult{SubmitStatus: "accepted"}, nil + } + pool := newMessageWorkPool(ctx, worker, 2) + if !pool.dispatch(message) { + t.Fatal("active message was not dispatched") + } + <-started + time.Sleep(5 * time.Millisecond) + if err := worker.recoverPending(ctx, pool); err != nil { + t.Fatalf("recoverPending: %v", err) + } + if got := submits.Load(); got != 1 { + t.Fatalf("active message submit count = %d, want 1", got) + } + close(release) + pool.wait() + if err := pool.dispatchRecovered(ctx, message); err != nil { + t.Fatalf("dispatch acknowledged recovery: %v", err) + } + if got := submits.Load(); got != 1 { + t.Fatalf("acknowledged message submit count = %d, want 1", got) } } diff --git a/gateway/internal/upstream/submit.go b/gateway/internal/upstream/submit.go index 5677b7c..94158a1 100644 --- a/gateway/internal/upstream/submit.go +++ b/gateway/internal/upstream/submit.go @@ -1,6 +1,7 @@ package upstream import ( + "cmpp-platform/gateway/internal/metrics" "cmpp-platform/gateway/internal/queue" "context" "fmt" @@ -17,7 +18,7 @@ import ( func (m *Manager) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.SubmitResult, error) { if err := validateSubmitCommand(cmd); err != nil { result := submitResult(cmd, 0, "", "rejected", "INVALID_COMMAND", err.Error()) - if postErr := m.post(ctx, "/gateway/events/submit-result", result); postErr != nil { + if postErr := m.postSubmitCallback(ctx, "/gateway/events/submit-result", result); postErr != nil { return result, postErr } return result, err @@ -26,7 +27,7 @@ func (m *Manager) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.Su pool, err := m.connectionFor(cmd) if err != nil { result := submitResult(cmd, 0, "", "rejected", "CONNECT_FAILED", err.Error()) - if postErr := m.post(ctx, "/gateway/events/submit-result", result); postErr != nil { + if postErr := m.postSubmitCallback(ctx, "/gateway/events/submit-result", result); postErr != nil { return result, postErr } return result, err @@ -43,7 +44,7 @@ func (m *Manager) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.Su SubmitSegmentResult: segment, } callbackCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) - postErr := m.post(callbackCtx, "/gateway/events/submit-segment-result", payload) + postErr := m.postSubmitCallback(callbackCtx, "/gateway/events/submit-segment-result", payload) cancel() if postErr != nil { log.Printf( @@ -53,12 +54,12 @@ func (m *Manager) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.Su } }) if err != nil { - if postErr := m.post(ctx, "/gateway/events/submit-result", result); postErr != nil { + if postErr := m.postSubmitCallback(ctx, "/gateway/events/submit-result", result); postErr != nil { return result, postErr } return result, err } - if err := m.post(ctx, "/gateway/events/submit-result", result); err != nil { + if err := m.postSubmitCallback(ctx, "/gateway/events/submit-result", result); err != nil { return result, err } return result, nil @@ -79,13 +80,17 @@ func (p *connectionPool) submit( var firstGatewayMessageID string segments := make([]queue.SubmitSegmentResult, 0, len(parts)) for _, part := range parts { + connectionStartedAt := time.Now() conn, release, err := p.acquireConnection(ctx) + metrics.ObserveSubmitStage("connection_wait", err == nil, time.Since(connectionStartedAt)) if err != nil { result := submitResult(cmd, 0, "", "timeout", "WINDOW_TIMEOUT", err.Error()) result.Segments = segments return result, err } + supplierStartedAt := time.Now() seq, gatewayMessageID, result, err := conn.submitPart(ctx, cmd, part) + metrics.ObserveSubmitStage("supplier_rtt", err == nil, time.Since(supplierStartedAt)) release() segment := submitSegmentResult(part, seq, gatewayMessageID, result) segments = append(segments, segment) @@ -113,6 +118,13 @@ func (p *connectionPool) submit( return result, nil } +func (m *Manager) postSubmitCallback(ctx context.Context, path string, payload any) error { + startedAt := time.Now() + err := m.post(ctx, path, payload) + metrics.ObserveSubmitStage("api_callback", err == nil, time.Since(startedAt)) + return err +} + func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, part submitPart) (uint32, string, queue.SubmitResult, error) { rspCh := make(chan submitPartResponse, 1) pkt := c.submitRequestPacket(cmd, part)