feat: 接入通知 outbox BullMQ 调度
- integrations/redis: 新增 BullMQ 连接参数工厂并复用 Redis 模块导出\n- notifications/wecom: 优先使用 BullMQ 周期调度 outbox 重试，Redis 不可用时降级进程内定时器\n- README/RTK/.env.example: 同步通知 outbox 重试和 Redis/BullMQ 配置说明
This commit is contained in:
湛兮
@@ -0,0 +1,33 @@
|
||||
/**
|
||||
* BullMQ 连接工厂统一封装 Redis URL 和 Worker 连接约束，避免业务模块散落连接参数。
|
||||
*/
|
||||
import { Injectable } from '@nestjs/common';
|
||||
import { ConfigService } from '@nestjs/config';
|
||||
import { ConnectionOptions } from 'bullmq';
|
||||
import { EnvConfig } from '../../config/env.schema';
|
||||
|
||||
@Injectable()
|
||||
export class BullMqConnectionFactory {
|
||||
constructor(private readonly config: ConfigService<EnvConfig, true>) {}
|
||||
|
||||
/**
|
||||
* 创建 BullMQ Queue/Worker 共用连接参数；Worker 阻塞连接要求请求重试不设上限。
|
||||
*/
|
||||
createConnectionOptions(
|
||||
connectionName: string,
|
||||
): ConnectionOptions | undefined {
|
||||
const redisUrl = this.config.get('REDIS_URL', { infer: true });
|
||||
|
||||
if (!redisUrl) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
return {
|
||||
url: redisUrl,
|
||||
connectionName,
|
||||
connectTimeout: 2000,
|
||||
maxRetriesPerRequest: null,
|
||||
retryStrategy: (attempt) => Math.min(attempt * 200, 5000),
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -1,8 +1,9 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
import { BullMqConnectionFactory } from './bullmq-connection.factory';
|
||||
import { RedisClient } from './redis.client';
|
||||
|
||||
@Module({
|
||||
providers: [RedisClient],
|
||||
exports: [RedisClient],
|
||||
providers: [BullMqConnectionFactory, RedisClient],
|
||||
exports: [BullMqConnectionFactory, RedisClient],
|
||||
})
|
||||
export class RedisModule {}
|
||||
|
||||
@@ -1,30 +1,52 @@
|
||||
/**
|
||||
* 通知 outbox 重试器复用当前 provider 投递边界，并避免空扫描刷屏审计。
|
||||
* 后续迁移到 BullMQ 时保留“有候选才记录成功 tick”的可观测边界。
|
||||
* 通知 outbox 重试器以 BullMQ 作为 Redis 调度层，并保留内存定时器降级路径。
|
||||
* 真实候选查询、投递和审计仍集中在 runOnce，避免队列层持有业务状态。
|
||||
*/
|
||||
import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
|
||||
import { ConfigService } from '@nestjs/config';
|
||||
import { Queue, Worker } from 'bullmq';
|
||||
import { AuditService } from '../../audit/audit.service';
|
||||
import { redactSensitive } from '../../common/security/redact-sensitive';
|
||||
import { EnvConfig } from '../../config/env.schema';
|
||||
import { BullMqConnectionFactory } from '../../integrations/redis/bullmq-connection.factory';
|
||||
import { NotificationOutboxService } from '../outbox/notification-outbox.service';
|
||||
import { DeployNotificationService } from './deploy-notification.service';
|
||||
|
||||
type NotificationOutboxRetryJobData = {
|
||||
source: 'bullmq';
|
||||
scheduledAt: string;
|
||||
};
|
||||
|
||||
const retryQueueName = 'notification-outbox-retry';
|
||||
const retryJobName = 'retry-tick';
|
||||
const retrySchedulerId = 'notification-outbox-retry-scheduler';
|
||||
const bullMqPrefix = 'devops-platform';
|
||||
const bullMqStartupTimeoutMs = 5000;
|
||||
|
||||
@Injectable()
|
||||
export class NotificationOutboxRetryService
|
||||
implements OnModuleInit, OnModuleDestroy
|
||||
{
|
||||
private timer: ReturnType<typeof setInterval> | undefined;
|
||||
private queue:
|
||||
| Queue<NotificationOutboxRetryJobData, void, string>
|
||||
| undefined;
|
||||
private worker:
|
||||
| Worker<NotificationOutboxRetryJobData, void, string>
|
||||
| undefined;
|
||||
private running = false;
|
||||
private bullMqFallbackRecorded = false;
|
||||
private bullMqRuntimeIssueRecorded = false;
|
||||
|
||||
constructor(
|
||||
private readonly config: ConfigService<EnvConfig, true>,
|
||||
private readonly bullMqConnectionFactory: BullMqConnectionFactory,
|
||||
private readonly notificationOutboxService: NotificationOutboxService,
|
||||
private readonly deployNotificationService: DeployNotificationService,
|
||||
private readonly auditService: AuditService,
|
||||
) {}
|
||||
|
||||
onModuleInit(): void {
|
||||
async onModuleInit(): Promise<void> {
|
||||
if (
|
||||
!this.config.get('NOTIFICATION_OUTBOX_RETRY_ENABLED', { infer: true })
|
||||
) {
|
||||
@@ -36,19 +58,120 @@ export class NotificationOutboxRetryService
|
||||
{ infer: true },
|
||||
);
|
||||
|
||||
await this.startBullMqScheduler(intervalMs);
|
||||
}
|
||||
|
||||
async onModuleDestroy(): Promise<void> {
|
||||
this.stopMemoryScheduler();
|
||||
await this.closeBullMqScheduler(false);
|
||||
}
|
||||
|
||||
private async startBullMqScheduler(intervalMs: number): Promise<void> {
|
||||
const connection = this.bullMqConnectionFactory.createConnectionOptions(
|
||||
'notification-outbox-retry',
|
||||
);
|
||||
|
||||
if (!connection) {
|
||||
this.startMemoryScheduler(intervalMs);
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
this.queue = new Queue<NotificationOutboxRetryJobData, void, string>(
|
||||
retryQueueName,
|
||||
{
|
||||
connection,
|
||||
prefix: bullMqPrefix,
|
||||
defaultJobOptions: {
|
||||
removeOnComplete: { count: 20 },
|
||||
removeOnFail: { count: 20 },
|
||||
},
|
||||
},
|
||||
);
|
||||
this.worker = new Worker<NotificationOutboxRetryJobData, void, string>(
|
||||
retryQueueName,
|
||||
async () => {
|
||||
await this.runOnce();
|
||||
},
|
||||
{
|
||||
connection,
|
||||
prefix: bullMqPrefix,
|
||||
concurrency: 1,
|
||||
removeOnComplete: { count: 20 },
|
||||
removeOnFail: { count: 20 },
|
||||
},
|
||||
);
|
||||
|
||||
this.queue.on('error', (error) => {
|
||||
void this.recordBullMqIssue('queue', error, 'bullmq_recovering');
|
||||
});
|
||||
this.worker.on('error', (error) => {
|
||||
void this.recordBullMqIssue('worker', error, 'bullmq_recovering');
|
||||
});
|
||||
this.worker.on('failed', (_job, error) => {
|
||||
void this.recordBullMqIssue('job', error, 'bullmq_recovering');
|
||||
});
|
||||
|
||||
await this.withStartupTimeout(
|
||||
Promise.all([
|
||||
this.queue.waitUntilReady(),
|
||||
this.worker.waitUntilReady(),
|
||||
]),
|
||||
);
|
||||
await this.queue.upsertJobScheduler(
|
||||
retrySchedulerId,
|
||||
{ every: intervalMs },
|
||||
{
|
||||
name: retryJobName,
|
||||
data: {
|
||||
source: 'bullmq',
|
||||
scheduledAt: new Date().toISOString(),
|
||||
},
|
||||
opts: {
|
||||
removeOnComplete: { count: 20 },
|
||||
removeOnFail: { count: 20 },
|
||||
},
|
||||
},
|
||||
);
|
||||
void this.runOnce();
|
||||
} catch (error) {
|
||||
await this.closeBullMqScheduler(true);
|
||||
await this.recordBullMqIssue('startup', error, 'memory_timer');
|
||||
this.startMemoryScheduler(intervalMs);
|
||||
}
|
||||
}
|
||||
|
||||
private startMemoryScheduler(intervalMs: number): void {
|
||||
if (this.timer) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.timer = setInterval(() => {
|
||||
void this.runOnce();
|
||||
}, intervalMs);
|
||||
void this.runOnce();
|
||||
}
|
||||
|
||||
onModuleDestroy(): void {
|
||||
private stopMemoryScheduler(): void {
|
||||
if (this.timer) {
|
||||
clearInterval(this.timer);
|
||||
this.timer = undefined;
|
||||
}
|
||||
}
|
||||
|
||||
private async closeBullMqScheduler(forceWorkerClose: boolean): Promise<void> {
|
||||
const worker = this.worker;
|
||||
const queue = this.queue;
|
||||
|
||||
this.worker = undefined;
|
||||
this.queue = undefined;
|
||||
|
||||
await Promise.all([
|
||||
worker?.close(forceWorkerClose).catch(() => undefined),
|
||||
queue?.close().catch(() => undefined),
|
||||
]);
|
||||
}
|
||||
|
||||
async runOnce(): Promise<void> {
|
||||
if (this.running) {
|
||||
return;
|
||||
@@ -63,13 +186,12 @@ export class NotificationOutboxRetryService
|
||||
const batchSize = this.config.get('NOTIFICATION_OUTBOX_BATCH_SIZE', {
|
||||
infer: true,
|
||||
});
|
||||
const messages = await this.notificationOutboxService.listRetryableMessages(
|
||||
{
|
||||
const messages =
|
||||
await this.notificationOutboxService.listRetryableMessages({
|
||||
now: new Date(),
|
||||
maxAttempts,
|
||||
take: batchSize,
|
||||
},
|
||||
);
|
||||
});
|
||||
let sentCount = 0;
|
||||
let failedCount = 0;
|
||||
let skippedCount = 0;
|
||||
@@ -116,4 +238,57 @@ export class NotificationOutboxRetryService
|
||||
this.running = false;
|
||||
}
|
||||
}
|
||||
|
||||
private async withStartupTimeout<T>(promise: Promise<T>): Promise<T> {
|
||||
return Promise.race([
|
||||
promise,
|
||||
new Promise<T>((_resolve, reject) => {
|
||||
setTimeout(
|
||||
() => reject(new Error('BullMQ 调度器启动超时')),
|
||||
bullMqStartupTimeoutMs,
|
||||
);
|
||||
}),
|
||||
]);
|
||||
}
|
||||
|
||||
private async recordBullMqIssue(
|
||||
stage: 'startup' | 'queue' | 'worker' | 'job',
|
||||
error: unknown,
|
||||
mode: 'memory_timer' | 'bullmq_recovering',
|
||||
): Promise<void> {
|
||||
const recorded =
|
||||
mode === 'memory_timer'
|
||||
? this.bullMqFallbackRecorded
|
||||
: this.bullMqRuntimeIssueRecorded;
|
||||
|
||||
if (recorded) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (mode === 'memory_timer') {
|
||||
this.bullMqFallbackRecorded = true;
|
||||
} else {
|
||||
this.bullMqRuntimeIssueRecorded = true;
|
||||
}
|
||||
|
||||
const message =
|
||||
error instanceof Error ? error.message : 'BullMQ 通知重试调度异常';
|
||||
|
||||
try {
|
||||
await this.auditService.record({
|
||||
action:
|
||||
mode === 'memory_timer'
|
||||
? 'NOTIFICATION_OUTBOX_RETRY_SCHEDULER_FALLBACK'
|
||||
: 'NOTIFICATION_OUTBOX_RETRY_SCHEDULER_DEGRADED',
|
||||
resourceType: 'notification_outbox',
|
||||
after: {
|
||||
stage,
|
||||
mode,
|
||||
message: redactSensitive({ message }).message,
|
||||
},
|
||||
});
|
||||
} catch {
|
||||
// 启动早期数据库可能尚不可用，调度降级不能被审计写入失败阻断。
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
* 通知模块集中注册 provider 客户端、发布通知服务和 outbox 能力。
|
||||
*/
|
||||
import { Module } from '@nestjs/common';
|
||||
import { RedisModule } from '../../integrations/redis/redis.module';
|
||||
import { NotificationOutboxModule } from '../outbox/notification-outbox.module';
|
||||
import { DeployNotificationService } from './deploy-notification.service';
|
||||
import { NotificationOutboxManualRetryController } from './notification-outbox-manual-retry.controller';
|
||||
@@ -10,7 +11,7 @@ import { NotificationOutboxRetryService } from './notification-outbox-retry.serv
|
||||
import { WeComClient } from './wecom.client';
|
||||
|
||||
@Module({
|
||||
imports: [NotificationOutboxModule],
|
||||
imports: [NotificationOutboxModule, RedisModule],
|
||||
controllers: [NotificationOutboxManualRetryController],
|
||||
providers: [
|
||||
WeComClient,
|
||||
|
||||
Reference in New Issue
Block a user