fix: harden dependencies and downstream delivery

2026-07-15 12:05:24 +08:00
parent 9bbfb72be6
commit a758672436
17 changed files with 418 additions and 57 deletions
@@ -1574,6 +1574,50 @@ describe('SendChainService', () => {
    });
  });

+  it('immediately terminates an unrecoverable downstream delivery', async () => {
+    const { service, prisma } = createService();
+    prisma.cmppDownstreamDelivery.findUnique.mockResolvedValue({
+      id: 'delivery-1',
+      tenantId: 'tenant-1',
+      applicationId: 'app-1',
+      messageId: 'MSG-1',
+      deliveryType: 'receipt',
+      retryCount: 0,
+      retryEnabled: true,
+    });
+
+    await service.markDownstreamDeliveryFailed(
+      'delivery-1',
+      '历史回执缺少原 Submit Sequence_Id，无法重建 Msg_Id',
+      'unrecoverable',
+    );
+
+    expect(prisma.cmppDownstreamDelivery.update).toHaveBeenCalledWith(expect.objectContaining({
+      data: expect.objectContaining({
+        status: 'failed',
+        retryCount: 1,
+        nextRetryAt: null,
+      }),
+    }));
+  });
+
+  it('does not let the pending timeout scan overwrite a delivery that is already awaiting acknowledgement', async () => {
+    const { service, prisma } = createService();
+    const awaitingAck = {
+      id: 'delivery-1', status: 'awaiting_ack', tenantId: 'tenant-1', applicationId: 'app-1',
+      messageId: 'MSG-1', deliveryType: 'receipt', retryCount: 0,
+    };
+    prisma.cmppDownstreamDelivery.findUnique.mockResolvedValue(awaitingAck);
+
+    await expect(service.markDownstreamDeliveryFailed(
+      'delivery-1',
+      '下游投递排队超过 72 小时，系统自动终止重试',
+      'queue_timeout',
+    )).resolves.toEqual(awaitingAck);
+
+    expect(prisma.cmppDownstreamDelivery.update).not.toHaveBeenCalled();
+  });
+
  it('only marks downstream delivery delivered after a successful CMPP_DELIVER_RESP', async () => {
    const { service, prisma } = createService();

@@ -1769,6 +1813,35 @@ describe('SendChainService', () => {
    expect(service['postGatewayControl']).not.toHaveBeenCalled();
  });

+  it('terminates a manual requeue when gateway reports it is unrecoverable', async () => {
+    const { service, prisma } = createService();
+    prisma.cmppDownstreamDelivery.findUnique.mockResolvedValueOnce({
+      id: 'delivery-1', tenantId: 'tenant-1', applicationId: 'app-1', messageId: 'MSG-1',
+      deliveryType: 'receipt', status: 'failed', retryCount: 3, manualRetryCount: 0,
+      payload: { account: '100001', messageId: 'MSG-1', receiptStatus: 'delivered' },
+      application: { cmppAccount: '100001' },
+    }).mockResolvedValueOnce({
+      id: 'delivery-1', tenantId: 'tenant-1', applicationId: 'app-1', messageId: 'MSG-1',
+      deliveryType: 'receipt', status: 'pending', retryCount: 0, retryEnabled: true,
+    });
+    service['postGatewayControl'] = jest.fn().mockResolvedValue({
+      sent: false,
+      retryable: false,
+      reasonCode: 'MISSING_SUBMIT_SEQUENCE_ID',
+      errorMessage: '历史回执缺少原 Submit Sequence_Id，无法重建 Msg_Id，系统已终止重投',
+    });
+
+    await service.requeueDownstreamDelivery('delivery-1');
+
+    expect(prisma.cmppDownstreamDelivery.update).toHaveBeenLastCalledWith(expect.objectContaining({
+      data: expect.objectContaining({
+        status: 'failed',
+        nextRetryAt: null,
+        lastError: expect.stringContaining('MISSING_SUBMIT_SEQUENCE_ID'),
+      }),
+    }));
+  });
+
  it('supports batch requeue of downstream deliveries', async () => {
    const { service } = createService();
    service.requeueDownstreamDelivery = jest.fn()
@@ -1819,16 +1892,41 @@ describe('SendChainService', () => {
    expect(prisma.smsBatchTask.update).toHaveBeenCalled();
  });

+  it('terminates downstream deliveries that remain pending for 72 hours after the latest manual retry', async () => {
+    const { service, prisma } = createService();
+    prisma.cmppDownstreamDelivery.findMany.mockResolvedValue([{ id: 'delivery-expired' }]);
+    service.markDownstreamDeliveryFailed = jest.fn().mockResolvedValue({ id: 'delivery-expired', status: 'failed' });
+
+    await expect(service.markExpiredDownstreamDeliveries(72)).resolves.toEqual({ failed: 1 });
+
+    expect(prisma.cmppDownstreamDelivery.findMany).toHaveBeenCalledWith(expect.objectContaining({
+      where: {
+        status: 'pending',
+        OR: [
+          { lastRetriedAt: null, createdAt: { lte: expect.any(Date) } },
+          { lastRetriedAt: { lte: expect.any(Date) } },
+        ],
+      },
+    }));
+    expect(service.markDownstreamDeliveryFailed).toHaveBeenCalledWith(
+      'delivery-expired',
+      '下游投递排队超过 72 小时，系统自动终止重试',
+      'queue_timeout',
+    );
+  });
+
  it('starts the automatic receipt-timeout scan after application startup', async () => {
    jest.useFakeTimers();
    const previousEnabled = process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED;
    const { service } = createService();
    const scan = jest.spyOn(service, 'markUnknownTimeout').mockResolvedValue({ timeout: 0 });
+    const downstreamScan = jest.spyOn(service, 'markExpiredDownstreamDeliveries').mockResolvedValue({ failed: 0 });
    try {
      process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED = 'true';
      service.onModuleInit();
      await jest.advanceTimersByTimeAsync(60_000);
      expect(scan).toHaveBeenCalledWith({});
+      expect(downstreamScan).toHaveBeenCalledWith();
      await service.onModuleDestroy();
    } finally {
      if (previousEnabled === undefined) delete process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED;
@@ -119,11 +119,21 @@ export interface GatewayDownstreamAcknowledgedDto extends GatewayDownstreamSentD
  acknowledgedAt?: string;
 }

-export type GatewayDownstreamFailureType = 'send_failed' | 'ack_timeout' | 'ack_rejected' | 'ack_invalid' | 'connection_lost';
+export type GatewayDownstreamFailureType =
+  | 'send_failed'
+  | 'ack_timeout'
+  | 'ack_rejected'
+  | 'ack_invalid'
+  | 'connection_lost'
+  | 'unrecoverable'
+  | 'queue_timeout';

 type GatewayControlDeliveryResult = {
  sent?: boolean;
  delivered?: boolean;
+  retryable?: boolean;
+  reasonCode?: string;
+  errorMessage?: string;
  connectionId?: string;
  sequenceId?: string;
  messageId?: string;
@@ -218,6 +228,7 @@ const GATEWAY_SUBMIT_STREAM = 'gateway.submit.commands';
 const DEFAULT_DOWNSTREAM_RETRY_DELAY_MS = 60_000;
 const DEFAULT_DOWNSTREAM_RETRY_MAX_DELAY_MS = 30 * 60_000;
 const DEFAULT_DOWNSTREAM_MAX_RETRIES = 10;
+const DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS = 72;
 const DEFAULT_RECEIPT_TIMEOUT_HOURS = 72;
 const DEFAULT_RECEIPT_TIMEOUT_SCAN_INTERVAL_MS = 5 * 60_000;
 const RECEIPT_TIMEOUT_INITIAL_DELAY_MS = 60_000;
@@ -879,6 +890,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
          receiptStatus: data.receiptStatus,
          rawStatus: data.rawStatus,
          errorCode: data.errorCode,
+          submitSequenceId: message.cmppSubmitSequenceId ? Number(message.cmppSubmitSequenceId) : undefined,
          deliveredAt: deliveredAt.toISOString(),
        },
      });
@@ -1044,10 +1056,14 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    if (delivery.status === 'delivered') {
      return delivery;
    }
+    if (failureType === 'queue_timeout' && delivery.status !== 'pending') {
+      return delivery;
+    }
    const retryCount = (delivery.retryCount ?? 0) + 1;
    const acknowledgementFailure = failureType === 'ack_timeout' || failureType === 'ack_rejected' || failureType === 'ack_invalid' || failureType === 'connection_lost';
    const retryAllowed = !acknowledgementFailure || delivery.retryEnabled !== false;
-    const finalFailure = !retryAllowed || retryCount >= downstreamMaxRetries();
+    const nonRetryableFailure = failureType === 'unrecoverable' || failureType === 'queue_timeout';
+    const finalFailure = nonRetryableFailure || !retryAllowed || retryCount >= downstreamMaxRetries();
    const finalStatus = failureType === 'ack_rejected' ? 'rejected' : acknowledgementFailure ? 'unconfirmed' : 'failed';
    const updated = await this.prisma.cmppDownstreamDelivery.update({
      where: { id },
@@ -1327,7 +1343,11 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      if (result.sent || result.delivered) {
        return this.markDownstreamDeliverySent({ id: delivery.id, ...result });
      }
-      return this.prisma.cmppDownstreamDelivery.findUnique({ where: { id: delivery.id } });
+      return this.markDownstreamDeliveryFailed(
+        delivery.id,
+        downstreamControlFailureMessage(result),
+        result.retryable === false ? 'unrecoverable' : 'send_failed',
+      );
    } catch (error) {
      return this.markDownstreamDeliveryFailed(
        delivery.id,
@@ -1504,6 +1524,12 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
      ) as GatewayControlDeliveryResult;
      if (result.sent || result.delivered) {
        await this.markDownstreamDeliverySent({ id: delivery.id, ...result });
+      } else {
+        await this.markDownstreamDeliveryFailed(
+          delivery.id,
+          downstreamControlFailureMessage(result),
+          result.retryable === false ? 'unrecoverable' : 'send_failed',
+        );
      }
    } catch (error) {
      await this.markDownstreamDeliveryFailed(delivery.id, error instanceof Error ? error.message : 'Gateway control delivery failed');
@@ -1884,12 +1910,39 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
    return { timeout };
  }

+  async markExpiredDownstreamDeliveries(olderThanHours = downstreamPendingTimeoutHours()) {
+    const cutoff = new Date(Date.now() - olderThanHours * 60 * 60_000);
+    const expired = await this.prisma.cmppDownstreamDelivery.findMany({
+      where: {
+        status: 'pending',
+        OR: [
+          { lastRetriedAt: null, createdAt: { lte: cutoff } },
+          { lastRetriedAt: { lte: cutoff } },
+        ],
+      },
+      select: { id: true },
+      take: 500,
+    });
+    for (const delivery of expired) {
+      await this.markDownstreamDeliveryFailed(
+        delivery.id,
+        `下游投递排队超过 ${olderThanHours} 小时，系统自动终止重试`,
+        'queue_timeout',
+      );
+    }
+    return { failed: expired.length };
+  }
+
  private async runReceiptTimeoutScan() {
    if (this.receiptTimeoutScanRunning) return;
    this.receiptTimeoutScanRunning = true;
    try {
-      const result = await this.markUnknownTimeout({});
-      if (result.timeout > 0) this.logger.log(`Marked ${result.timeout} messages as receipt timeout and refunded charged messages`);
+      const [receiptResult, downstreamResult] = await Promise.all([
+        this.markUnknownTimeout({}),
+        this.markExpiredDownstreamDeliveries(),
+      ]);
+      if (receiptResult.timeout > 0) this.logger.log(`Marked ${receiptResult.timeout} messages as receipt timeout and refunded charged messages`);
+      if (downstreamResult.failed > 0) this.logger.log(`Terminated ${downstreamResult.failed} expired downstream deliveries`);
    } catch (error) {
      this.logger.error('Receipt timeout scan failed', error instanceof Error ? error.stack : String(error));
    } finally {
@@ -2863,6 +2916,20 @@ function downstreamMaxRetries() {
  return Number.isFinite(value) && value > 0 ? Math.floor(value) : DEFAULT_DOWNSTREAM_MAX_RETRIES;
 }

+function downstreamPendingTimeoutHours() {
+  const value = Number(process.env.CMPP_DOWNSTREAM_PENDING_TIMEOUT_HOURS ?? DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS);
+  return Number.isFinite(value) && value > 0 ? value : DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS;
+}
+
+function downstreamControlFailureMessage(result: GatewayControlDeliveryResult) {
+  const reason = String(result.errorMessage ?? '').trim();
+  const code = String(result.reasonCode ?? '').trim();
+  if (reason && code) return `${reason} (${code})`;
+  if (reason) return reason;
+  if (code) return `Gateway 未完成下游投递 (${code})`;
+  return 'Gateway 未完成下游投递，等待自动重试';
+}
+
 function parseImportRows(content: string, delimiter?: ',' | '\t') {
  const normalized = content.replace(/^\uFEFF/, '');
  const lines = normalized.split(/\r?\n/).filter((line) => line.trim().length > 0);