From 14e2ec92fd214cb2469a9f8ad8789b38e2bc479e Mon Sep 17 00:00:00 2001 From: sunlei Date: Tue, 16 Jun 2026 17:10:48 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8DQQBot=E6=8F=92?= =?UTF-8?q?=E4=BB=B6=E9=98=9F=E5=88=97=E6=8E=92=E9=98=9F=E8=B6=85=E6=97=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 1 + API.md | 4 +- README.md | 4 +- .../bullmq-plugin-worker-request.queue.ts | 51 ++++++++++++++++++- .../integration/runtime/worker-runtime.ts | 13 ++++- .../runtime/worker-runtime.types.ts | 2 + .../bullmq-worker-queue.spec.ts | 3 ++ .../plugin-platform/worker-runtime.spec.ts | 49 ++++++++++++++++++ 8 files changed, 120 insertions(+), 7 deletions(-) diff --git a/.env.example b/.env.example index 3c788b5..e4bf597 100644 --- a/.env.example +++ b/.env.example @@ -63,6 +63,7 @@ QQBOT_PLUGIN_QUEUE_REDIS_DB=0 QQBOT_PLUGIN_QUEUE_REDIS_PREFIX=kt:qqbot:plugin-worker QQBOT_PLUGIN_QUEUE_REMOVE_ON_FAIL=100 QQBOT_PLUGIN_QUEUE_WAIT_BUFFER_MS=5000 +QQBOT_PLUGIN_QUEUE_WAIT_TIMEOUT_MS=120000 QQBOT_COMMAND_MIN_COOLDOWN_MS=5000 QQBOT_RULE_MIN_COOLDOWN_MS=30000 QQBOT_REPEATER_THRESHOLD=4 diff --git a/API.md b/API.md index 92cd107..b907573 100644 --- a/API.md +++ b/API.md @@ -85,7 +85,7 @@ Admin、Component、Dict、MinIO、Blog 管理、WordPress 管理和 QQBot 管 | Admin | `ADMIN_TOKEN_SECRET`、`ADMIN_COOKIE_SECURE`、`SNOWFLAKE_WORKER_ID`、`SNOWFLAKE_DATACENTER_ID` | | WordPress | `WORDPRESS_BASE_URL`、`WORDPRESS_HOST_HEADER`、`WORDPRESS_ADMIN_USERNAME`、`WORDPRESS_ADMIN_PASSWORD` | | Loki | `LOG_LEVEL`、`LOG_APP_NAME`、`LOKI_URL`、`LOKI_QUERY_HOST`、`LOKI_QUERY_SELECTOR` | -| QQBot | `QQBOT_ENABLED`、`QQBOT_ACCOUNT_SECRET_KEY`、`QQBOT_REVERSE_WS_PATH`、`QQBOT_REVERSE_WS_TOKEN`、`QQBOT_EVENT_BUS`、`QQBOT_SEND_*`、`QQBOT_PLUGIN_QUEUE_REDIS_*`、`QQBOT_COMMAND_MIN_COOLDOWN_MS`、`QQBOT_RULE_MIN_COOLDOWN_MS`、`QQBOT_REPEATER_*` | +| QQBot | `QQBOT_ENABLED`、`QQBOT_ACCOUNT_SECRET_KEY`、`QQBOT_REVERSE_WS_PATH`、`QQBOT_REVERSE_WS_TOKEN`、`QQBOT_EVENT_BUS`、`QQBOT_SEND_*`、`QQBOT_PLUGIN_QUEUE_REDIS_*`、`QQBOT_PLUGIN_QUEUE_WAIT_TIMEOUT_MS`、`QQBOT_COMMAND_MIN_COOLDOWN_MS`、`QQBOT_RULE_MIN_COOLDOWN_MS`、`QQBOT_REPEATER_*` | | NapCat | `NAPCAT_WEBUI_BASE_URL`、`NAPCAT_WEBUI_TOKEN`、`QQBOT_NAPCAT_*` | | MQTT | `MQTT_URL`、`MQTT_USERNAME`、`MQTT_PASSWORD`、`MQTT_CLIENT_ID` | | BangDream | `BANGDREAM_TSUGU_MAIN_SERVER`、`BANGDREAM_TSUGU_DISPLAYED_SERVERS`、`BANGDREAM_TSUGU_CACHE_ROOT` | @@ -94,7 +94,7 @@ Admin、Component、Dict、MinIO、Blog 管理、WordPress 管理和 QQBot 管 真实密码、Token、OAuth secret 和生产 env 不提交到 Git。 -QQBot 插件 worker 队列依赖 Redis。K8s 生产清单提供内部 Redis Service `kt-qqbot-plugin-redis:6379`，用于 `QQBOT_PLUGIN_QUEUE_REDIS_HOST` / `QQBOT_PLUGIN_QUEUE_REDIS_PORT`。 +QQBot 插件 worker 队列依赖 Redis。K8s 生产清单提供内部 Redis Service `kt-qqbot-plugin-redis:6379`，用于 `QQBOT_PLUGIN_QUEUE_REDIS_HOST` / `QQBOT_PLUGIN_QUEUE_REDIS_PORT`。`QQBOT_PLUGIN_QUEUE_WAIT_TIMEOUT_MS` 控制串行队列的排队等待窗口，避免排队时间挤占插件操作本身的 `timeoutMs` 执行预算。 ## Admin 与基础后台 diff --git a/README.md b/README.md index df829d5..fbbae77 100644 --- a/README.md +++ b/README.md @@ -62,14 +62,14 @@ ci/ Jenkins Agent/Docker 辅助文件 | Admin | `ADMIN_TOKEN_SECRET`、`ADMIN_COOKIE_SECURE`、`SNOWFLAKE_WORKER_ID`、`SNOWFLAKE_DATACENTER_ID` | | WordPress | `WORDPRESS_BASE_URL`、`WORDPRESS_HOST_HEADER`、`WORDPRESS_ADMIN_USERNAME`、`WORDPRESS_ADMIN_PASSWORD`、`WORDPRESS_*_TIMEOUT_MS` | | Logging/Loki | `LOG_LEVEL`、`LOG_APP_NAME`、`LOKI_URL`、`LOKI_QUERY_HOST`、`LOKI_*` | -| QQBot/NapCat | `QQBOT_ENABLED`、`QQBOT_ACCOUNT_SECRET_KEY`、`QQBOT_REVERSE_WS_*`、`QQBOT_SEND_*`、`QQBOT_PLUGIN_QUEUE_REDIS_*`、`QQBOT_COMMAND_MIN_COOLDOWN_MS`、`QQBOT_RULE_MIN_COOLDOWN_MS`、`QQBOT_REPEATER_*`、`NAPCAT_*`、`QQBOT_NAPCAT_*`、`MQTT_*` | +| QQBot/NapCat | `QQBOT_ENABLED`、`QQBOT_ACCOUNT_SECRET_KEY`、`QQBOT_REVERSE_WS_*`、`QQBOT_SEND_*`、`QQBOT_PLUGIN_QUEUE_REDIS_*`、`QQBOT_PLUGIN_QUEUE_WAIT_TIMEOUT_MS`、`QQBOT_COMMAND_MIN_COOLDOWN_MS`、`QQBOT_RULE_MIN_COOLDOWN_MS`、`QQBOT_REPEATER_*`、`NAPCAT_*`、`QQBOT_NAPCAT_*`、`MQTT_*` | | BangDream | `BANGDREAM_TSUGU_MAIN_SERVER`、`BANGDREAM_TSUGU_DISPLAYED_SERVERS`、`BANGDREAM_TSUGU_CACHE_ROOT` | | FF14 Market | `FF14_XIVAPI_BASE_URL`、`FF14_UNIVERSALIS_BASE_URL`、`FF14_MARKET_CACHE_TTL_MS` | | FFLogs | `FFLOGS_BASE_URL`、`FFLOGS_GRAPHQL_URL`、`FFLOGS_TOKEN_URL`、`FFLOGS_CLIENT_ID`、`FFLOGS_CLIENT_SECRET` | `DB_SYNC=true` 只适合本地开发或明确允许自动同步表结构的环境；生产应关闭并使用 SQL/迁移脚本。 -QQBot 插件 worker 使用 BullMQ 队列串行执行同一插件安装实例的请求。K8s 生产清单包含内部服务 `kt-qqbot-plugin-redis`，生产 env 可将 `QQBOT_PLUGIN_QUEUE_REDIS_HOST` 配为该服务名。 +QQBot 插件 worker 使用 BullMQ 队列串行执行同一插件安装实例的请求。K8s 生产清单包含内部服务 `kt-qqbot-plugin-redis`，生产 env 可将 `QQBOT_PLUGIN_QUEUE_REDIS_HOST` 配为该服务名。`QQBOT_PLUGIN_QUEUE_WAIT_TIMEOUT_MS` 控制排队等待窗口，插件 `operation.timeoutMs` 仍表示单次执行预算。 ## 启动 diff --git a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/bullmq-plugin-worker-request.queue.ts b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/bullmq-plugin-worker-request.queue.ts index 1686131..2e9f539 100644 --- a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/bullmq-plugin-worker-request.queue.ts +++ b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/bullmq-plugin-worker-request.queue.ts @@ -25,6 +25,7 @@ export type QqbotBullmqWorkerRequestQueueOptions = { installationId: string; pluginKey: string; prefix: string; + queueWaitTimeoutMs: number; removeOnFailCount: number; waitUntilFinishedBufferMs: number; workerInstanceId: string; @@ -48,6 +49,8 @@ type QqbotWorkerQueueResult = }; export class QqbotBullmqPluginWorkerRequestQueue implements QqbotPluginWorkerRequestQueue { + readonly handlesRequestTimeout = true; + readonly queueWaitTimeoutMs: number; private readonly queue: Queue< QqbotWorkerQueueJobData, QqbotWorkerQueueResult, @@ -70,6 +73,7 @@ export class QqbotBullmqPluginWorkerRequestQueue implements QqbotPluginWorkerReq private readonly driver: QqbotPluginWorkerDriver, private readonly options: QqbotBullmqWorkerRequestQueueOptions, ) { + this.queueWaitTimeoutMs = options.queueWaitTimeoutMs; const queueName = buildWorkerQueueName( options.pluginKey, options.installationId, @@ -113,7 +117,7 @@ export class QqbotBullmqPluginWorkerRequestQueue implements QqbotPluginWorkerReq try { return { ok: true, - result: await this.driver.request(job.data.message), + result: await this.requestDriverWithTimeout(job.data.message), }; } catch (error) { if (error instanceof QqbotPluginWorkerResponseError) { @@ -122,6 +126,16 @@ export class QqbotBullmqPluginWorkerRequestQueue implements QqbotPluginWorkerReq ok: false, }; } + if (error instanceof QqbotPluginWorkerExpiredRequestError) { + await this.reset(); + return { + error: { + message: error.message, + name: error.name, + }, + ok: false, + }; + } await this.reset(); throw error; } @@ -159,6 +173,7 @@ export class QqbotBullmqPluginWorkerRequestQueue implements QqbotPluginWorkerReq expiresAt: Date.now() + message.timeoutMs + + this.options.queueWaitTimeoutMs + this.options.waitUntilFinishedBufferMs, generation: this.generation, message, @@ -173,7 +188,9 @@ export class QqbotBullmqPluginWorkerRequestQueue implements QqbotPluginWorkerReq ); const result = await job.waitUntilFinished( this.queueEvents, - message.timeoutMs + this.options.waitUntilFinishedBufferMs, + message.timeoutMs + + this.options.queueWaitTimeoutMs + + this.options.waitUntilFinishedBufferMs, ); if (result?.ok === false) { @@ -212,6 +229,31 @@ export class QqbotBullmqPluginWorkerRequestQueue implements QqbotPluginWorkerReq error.stack, ); } + + private async requestDriverWithTimeout( + message: QqbotPluginWorkerRequest, + ): Promise { + let timer: NodeJS.Timeout | undefined; + const requestPromise = this.driver.request(message); + requestPromise.catch(() => undefined); + + const timeoutPromise = new Promise((_, reject) => { + timer = setTimeout(() => { + reject( + new QqbotPluginWorkerExpiredRequestError( + 'worker-request-execution-timeout', + ), + ); + }, message.timeoutMs); + timer.unref?.(); + }); + + try { + return await Promise.race([requestPromise, timeoutPromise]); + } finally { + if (timer) clearTimeout(timer); + } + } } export function createQqbotBullmqWorkerQueueOptions( @@ -224,6 +266,11 @@ export function createQqbotBullmqWorkerQueueOptions( installationId, pluginKey, prefix: resolveQqbotPluginQueuePrefix(configService), + queueWaitTimeoutMs: readNumberConfig( + configService, + ['QQBOT_PLUGIN_QUEUE_WAIT_TIMEOUT_MS'], + 120_000, + ), removeOnFailCount: readNumberConfig( configService, ['QQBOT_PLUGIN_QUEUE_REMOVE_ON_FAIL'], diff --git a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.ts b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.ts index 7186a61..7237acf 100644 --- a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.ts +++ b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.ts @@ -188,7 +188,11 @@ export class QqbotPluginWorkerRuntime { ); requestPromise.catch(() => undefined); - const timeout = this.createTimeoutPromise(type, message, timeoutMs); + const timeout = this.createTimeoutPromise( + type, + message, + this.getRequestTimeoutMs(timeoutMs), + ); try { return await Promise.race([requestPromise, timeout.promise]); @@ -287,6 +291,13 @@ export class QqbotPluginWorkerRuntime { }; } + private getRequestTimeoutMs(timeoutMs: number) { + if (!this.requestQueue.handlesRequestTimeout) return timeoutMs; + + const queueWaitTimeoutMs = Number(this.requestQueue.queueWaitTimeoutMs || 0); + return timeoutMs + Math.max(0, queueWaitTimeoutMs); + } + private async recoverIfNeeded(triggerType: QqbotPluginWorkerRequestType) { if (this.status !== 'failed' || !this.manifestForRecovery) return; diff --git a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.types.ts b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.types.ts index ba21ad4..e632251 100644 --- a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.types.ts +++ b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.types.ts @@ -40,6 +40,8 @@ export type QqbotPluginWorkerDriver = { export type QqbotPluginWorkerRequestQueue = { close(): Promise; + handlesRequestTimeout?: boolean; + queueWaitTimeoutMs?: number; request(message: QqbotPluginWorkerRequest): Promise; reset(): Promise; }; diff --git a/test/modules/qqbot/plugin-platform/bullmq-worker-queue.spec.ts b/test/modules/qqbot/plugin-platform/bullmq-worker-queue.spec.ts index 0b24482..9ae04dd 100644 --- a/test/modules/qqbot/plugin-platform/bullmq-worker-queue.spec.ts +++ b/test/modules/qqbot/plugin-platform/bullmq-worker-queue.spec.ts @@ -74,6 +74,7 @@ describe('QQBot BullMQ plugin worker request queue', () => { installationId: 'install:1', pluginKey: 'bang:dream', prefix: 'kt:qqbot:plugin-worker', + queueWaitTimeoutMs: 120_000, removeOnFailCount: 100, waitUntilFinishedBufferMs: 5_000, workerInstanceId: 'worker-1', @@ -95,5 +96,7 @@ describe('QQBot BullMQ plugin worker request queue', () => { (creation) => creation.options.prefix === 'kt:qqbot:plugin-worker', ), ).toBe(true); + expect(queue.handlesRequestTimeout).toBe(true); + expect(queue.queueWaitTimeoutMs).toBe(120_000); }); }); diff --git a/test/modules/qqbot/plugin-platform/worker-runtime.spec.ts b/test/modules/qqbot/plugin-platform/worker-runtime.spec.ts index 2da7b18..2addde9 100644 --- a/test/modules/qqbot/plugin-platform/worker-runtime.spec.ts +++ b/test/modules/qqbot/plugin-platform/worker-runtime.spec.ts @@ -106,6 +106,11 @@ class GenerationAwareRequestQueue implements QqbotPluginWorkerRequestQueue { } } +class TimeoutAwareRecordingRequestQueue extends RecordingRequestQueue { + readonly handlesRequestTimeout = true; + readonly queueWaitTimeoutMs = 50; +} + const createRuntime = (driver = new RecordingDriver()) => { const runtime = new QqbotPluginWorkerRuntime(new RecordingRequestQueue(driver), { defaultTimeoutMs: 50, @@ -179,6 +184,49 @@ describe('QQBot plugin worker runtime', () => { } }); + it('does not count queue wait time against a queued operation execution timeout', async () => { + const requestOrder: string[] = []; + const driver: QqbotPluginWorkerDriver = { + dispose: jest.fn(async () => undefined), + request: jest.fn(async (message) => { + requestOrder.push(message.operationId || message.type); + await new Promise((resolve) => + setTimeout(resolve, message.operationId === 'op-1' ? 20 : 15), + ); + return { + ok: true, + operationId: message.operationId, + }; + }), + }; + const runtime = new QqbotPluginWorkerRuntime( + new TimeoutAwareRecordingRequestQueue(driver), + { + defaultTimeoutMs: 30, + installationId: 'install-queue-wait', + pluginKey: 'demo-plugin', + }, + ); + + const first = runtime.executeOperation({ + input: { text: 'first' }, + operationId: 'op-1', + operationKey: 'demo-plugin.echo', + }); + const second = runtime.executeOperation({ + input: { text: 'second' }, + operationId: 'op-2', + operationKey: 'demo-plugin.echo', + }); + + await expect(Promise.all([first, second])).resolves.toEqual([ + { ok: true, operationId: 'op-1' }, + { ok: true, operationId: 'op-2' }, + ]); + expect(requestOrder).toEqual(['op-1', 'op-2']); + expect(driver.dispose).not.toHaveBeenCalled(); + }); + it('recovers a failed worker by reloading and activating it before the next request', async () => { const requestTypes: string[] = []; const driver: QqbotPluginWorkerDriver = { @@ -596,6 +644,7 @@ describe('QQBot plugin worker queue config', () => { installationId: 'install-1', pluginKey: 'bangdream', prefix: 'kt:qqbot:plugin-worker', + queueWaitTimeoutMs: 120_000, removeOnFailCount: 100, waitUntilFinishedBufferMs: 5_000, workerInstanceId: expect.any(String),