KTeam/kt-template-online-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 修复QQBot插件队列排队超时

Browse Source
This commit is contained in:
sunlei 2026-06-16 17:10:48 +08:00
parent 5926f50fa3
commit 14e2ec92fd
8 changed files with 120 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.env.example
View File

@ -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

4
API.md
View File

@ -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 与基础后台

4
README.md
View File

@ -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` 仍表示单次执行预算。
## 启动

51
src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/bullmq-plugin-worker-request.queue.ts
View File

@ -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<unknown> {
let timer: NodeJS.Timeout | undefined;
const requestPromise = this.driver.request(message);
requestPromise.catch(() => undefined);
const timeoutPromise = new Promise<never>((_, 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'],

13
src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.ts
View File

@ -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;

2
src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.types.ts
View File

@ -40,6 +40,8 @@ export type QqbotPluginWorkerDriver = {
export type QqbotPluginWorkerRequestQueue = {
close(): Promise<void>;
handlesRequestTimeout?: boolean;
queueWaitTimeoutMs?: number;
request(message: QqbotPluginWorkerRequest): Promise<unknown>;
reset(): Promise<void>;
};

3
test/modules/qqbot/plugin-platform/bullmq-worker-queue.spec.ts
View File

@ -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);
});
});

49
test/modules/qqbot/plugin-platform/worker-runtime.spec.ts
View File

@ -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),