From d74545d17b6a98adb53b21a873c3905a6bb153e3 Mon Sep 17 00:00:00 2001 From: sunlei Date: Fri, 24 Jul 2026 13:32:11 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=94=B6=E7=B4=A7=E7=B3=BB=E7=BB=9F?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E6=8A=95=E9=80=92=E8=BE=B9=E7=95=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../account/qqbot-account.service.ts | 12 +- .../qqbot-message-subscription.service.ts | 28 +++- .../system-message-delivery-runner.service.ts | 75 +++++++---- .../core/application/send/qqbot-send.error.ts | 19 ++- .../application/send/qqbot-send.service.ts | 17 +-- ...qqbot-message-subscription.service.spec.ts | 17 ++- .../message-push/qqbot-strict-send.spec.ts | 56 +++++++- .../system-message-delivery-runner.spec.ts | 121 +++++++++++++++++- 8 files changed, 288 insertions(+), 57 deletions(-) diff --git a/src/modules/qqbot/core/application/account/qqbot-account.service.ts b/src/modules/qqbot/core/application/account/qqbot-account.service.ts index 0dbe731..f0af644 100644 --- a/src/modules/qqbot/core/application/account/qqbot-account.service.ts +++ b/src/modules/qqbot/core/application/account/qqbot-account.service.ts @@ -392,8 +392,10 @@ export class QqbotAccountService { } /** - * 更新数据。 - * @param body - 请求体 DTO；承载 QQBot新增、更新、导入或执行字段。 + * Updates the account and ability identity atomically, cancelling its claimable deliveries + * in the same transaction when an administrator disables it or replaces its self ID. + * @param body - Complete administrative account update with the target account ID. + * @returns `true` after the account, ability, and required delivery cancellations commit. */ async update(body: QqbotAccountUpdateDto) { if (body.selfId) { @@ -429,8 +431,10 @@ export class QqbotAccountService { } /** - * 删除数据。 - * @param id - QQBot记录 ID；定位本次读取、更新、删除或关联的QQBot记录。 + * Removes NapCat containers first, then soft-deletes the account and ability and cancels + * claimable deliveries in one database transaction; a later DB failure cannot restore containers. + * @param id - QQBot account ID to remove. + * @returns `{ deletedContainers: number }` for containers removed before the transaction. */ async remove(id: string) { const account = await this.accountRepository.findOne({ diff --git a/src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts b/src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts index d5c91c9..8ac1304 100644 --- a/src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts +++ b/src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts @@ -172,9 +172,13 @@ export class QqbotMessageSubscriptionService { Object.assign(current, normalized); const saved = await repository.save(current); if (!saved.enabled || sourceIdentityChanged) { - await this.cancelUnfinishedDeliveries(manager, { - subscriptionId: saved.id, - }); + await this.cancelUnfinishedDeliveries( + manager, + { + subscriptionId: saved.id, + }, + sourceIdentityChanged, + ); } return saved; }, @@ -365,13 +369,27 @@ export class QqbotMessageSubscriptionService { } } - /** Cancels only claimable historical deliveries in the caller's configuration transaction. */ + /** + * Cancels historical deliveries in the caller's configuration transaction. + * @param manager - Transaction manager that already owns the subscription mutation. + * @param where - Exact subscription identity whose historical rows are invalidated. + * @param includeProcessing - Whether a canonical identity change must revoke active owners too. + */ private async cancelUnfinishedDeliveries( manager: EntityManager, where: Pick, + includeProcessing = false, ): Promise { await manager.getRepository(QqbotMessageDelivery).update( - { ...where, status: In(['waiting_ddns', 'pending', 'retry']) }, + { + ...where, + status: In([ + 'waiting_ddns', + 'pending', + 'retry', + ...(includeProcessing ? (['processing'] as const) : []), + ]), + }, { status: 'cancelled', nextAttemptAt: null, diff --git a/src/modules/qqbot/core/application/message-push/system-message-delivery-runner.service.ts b/src/modules/qqbot/core/application/message-push/system-message-delivery-runner.service.ts index 5ea1d01..4e62ac0 100644 --- a/src/modules/qqbot/core/application/message-push/system-message-delivery-runner.service.ts +++ b/src/modules/qqbot/core/application/message-push/system-message-delivery-runner.service.ts @@ -11,7 +11,10 @@ import { QqbotMessageEvent } from '../../infrastructure/persistence/message-push import { QqbotMessagePublishBinding } from '../../infrastructure/persistence/message-push/qqbot-message-publish-binding.entity'; import { QqbotMessagePublishTarget } from '../../infrastructure/persistence/message-push/qqbot-message-publish-target.entity'; import { QqbotMessageSubscription } from '../../infrastructure/persistence/message-push/qqbot-message-subscription.entity'; -import { QqbotSendAttemptError } from '../send/qqbot-send.error'; +import { + QqbotSendAttemptError, + strictSendErrorSummary, +} from '../send/qqbot-send.error'; import { QqbotSendService } from '../send/qqbot-send.service'; import { SYSTEM_MESSAGE_BATCH_SIZE, @@ -61,16 +64,16 @@ export class SystemMessageDeliveryRunnerService { ) {} /** Processes up to one bounded batch of due deliveries and returns its claim count. */ - async runOnce(now: Date = new Date()): Promise { + async runOnce(now?: Date): Promise { let claimed = 0; for (let index = 0; index < SYSTEM_MESSAGE_BATCH_SIZE; index += 1) { - const token = await this.claimOne(now); + const token = await this.claimOne(now ?? new Date()); if (!token) break; claimed += 1; try { await this.processClaim(token, now); } catch { - await this.handleUnexpectedClaimFailure(token, now); + await this.handleUnexpectedClaimFailure(token, now ?? new Date()); } } return claimed; @@ -115,8 +118,12 @@ export class SystemMessageDeliveryRunnerService { } /** Rechecks one claimed delivery before doing external I/O and then performs its final CAS. */ - private async processClaim(token: ClaimToken, now: Date): Promise { - if (now.getTime() >= token.delivery.expiresAt.getTime()) { + private async processClaim( + token: ClaimToken, + fixedNow?: Date, + ): Promise { + const preparationNow = fixedNow ?? new Date(); + if (preparationNow.getTime() >= token.delivery.expiresAt.getTime()) { await this.finish( token, 'failed', @@ -126,17 +133,33 @@ export class SystemMessageDeliveryRunnerService { ); return; } - const prepared = await this.prepare(token, now); + const prepared = await this.prepare(token); if (prepared.kind === 'stale') return; if (prepared.kind === 'finish') { if (prepared.status === 'waiting_ddns') { + const schedulingNow = fixedNow ?? new Date(); + if ( + schedulingNow.getTime() + SYSTEM_MESSAGE_DDNS_RECHECK_MS >= + token.delivery.expiresAt.getTime() + ) { + await this.finish( + token, + 'failed', + DELIVERY_EXPIRED, + 'delivery deadline reached', + null, + ); + return; + } await this.finish( token, 'waiting_ddns', prepared.code, 'DDNS is not synchronized', null, - new KtDateTime(now.getTime() + SYSTEM_MESSAGE_DDNS_RECHECK_MS), + new KtDateTime( + schedulingNow.getTime() + SYSTEM_MESSAGE_DDNS_RECHECK_MS, + ), ); } else { await this.finish( @@ -149,6 +172,17 @@ export class SystemMessageDeliveryRunnerService { } return; } + const sendNow = fixedNow ?? new Date(); + if (sendNow.getTime() >= token.delivery.expiresAt.getTime()) { + await this.finish( + token, + 'failed', + DELIVERY_EXPIRED, + 'delivery deadline reached', + null, + ); + return; + } try { const result = await this.sendService.sendStrictPlainText({ attemptNumber: token.attempt, @@ -166,23 +200,23 @@ export class SystemMessageDeliveryRunnerService { token, 'failed', error.code, - this.safeMessage(error), + strictSendErrorSummary(error.code), error.sendLogId, ); return; } await this.retryOrFail( token, - now, + fixedNow ?? new Date(), error.code, - this.safeMessage(error), + strictSendErrorSummary(error.code), error.sendLogId, ); return; } await this.retryOrFail( token, - now, + fixedNow ?? new Date(), TRANSIENT_ERROR, 'delivery transport unavailable', null, @@ -191,10 +225,7 @@ export class SystemMessageDeliveryRunnerService { } /** Locks current configuration before locking the delivery and returns a safe next action. */ - private async prepare( - token: ClaimToken, - now: Date, - ): Promise { + private async prepare(token: ClaimToken): Promise { return this.dataSource.transaction(async (manager) => { const event = await manager.getRepository(QqbotMessageEvent).findOne({ where: { id: token.delivery.messageEventId }, @@ -285,11 +316,6 @@ export class SystemMessageDeliveryRunnerService { subscriptionConfig: subscription.sourceConfig, }); if (readiness.status === 'waiting_ddns') { - if ( - now.getTime() + SYSTEM_MESSAGE_DDNS_RECHECK_MS >= - delivery.expiresAt.getTime() - ) - return { code: DELIVERY_EXPIRED, kind: 'finish', status: 'failed' }; return { code: readiness.reasonCode, kind: 'finish', @@ -428,11 +454,4 @@ export class SystemMessageDeliveryRunnerService { delivery.processingLeaseUntil?.getTime() === token.leaseUntil.getTime() ); } - - /** Converts unexpected error text into a bounded persistence-safe summary. */ - private safeMessage(error: Error): string { - return String(error.message || 'delivery failed') - .replace(/[\r\n]/g, ' ') - .slice(0, 500); - } } diff --git a/src/modules/qqbot/core/application/send/qqbot-send.error.ts b/src/modules/qqbot/core/application/send/qqbot-send.error.ts index a2d90db..1d4df4e 100644 --- a/src/modules/qqbot/core/application/send/qqbot-send.error.ts +++ b/src/modules/qqbot/core/application/send/qqbot-send.error.ts @@ -1,5 +1,18 @@ import type { QqbotSendAttemptErrorOptions } from '../../contract/message-push/qqbot-message-push.types'; +const STRICT_SEND_ERROR_SUMMARIES: Readonly> = { + account_unavailable: 'Configured QQBot account is unavailable', + invalid_target_type: 'Strict QQBot delivery target type is invalid', + onebot_disconnected: 'OneBot connection unavailable', + onebot_rejected: 'OneBot rejected the send action', + onebot_timeout: 'OneBot send timed out', +}; + +/** Maps a strict-send classification to its allowlisted non-sensitive summary. */ +export function strictSendErrorSummary(code: string): string { + return STRICT_SEND_ERROR_SUMMARIES[code] ?? 'QQBot delivery failed'; +} + /** Represents a retryable or permanent strict QQBot delivery attempt failure. */ export class QqbotSendAttemptError extends Error { readonly code: string; @@ -7,11 +20,11 @@ export class QqbotSendAttemptError extends Error { readonly sendLogId: null | string; /** - * Preserves the stable retry classification and optional pending send-log identity. - * @param options - The approved code, non-sensitive message, retry policy, and log ID. + * Preserves stable retry metadata while replacing all caller text with an allowlisted summary. + * @param options - The stable code, ignored raw message, retry policy, and optional log ID. */ constructor(options: QqbotSendAttemptErrorOptions) { - super(options.message); + super(strictSendErrorSummary(options.code)); this.name = 'QqbotSendAttemptError'; this.code = options.code; this.retryable = options.retryable; diff --git a/src/modules/qqbot/core/application/send/qqbot-send.service.ts b/src/modules/qqbot/core/application/send/qqbot-send.service.ts index 0162d9b..28c6416 100644 --- a/src/modules/qqbot/core/application/send/qqbot-send.service.ts +++ b/src/modules/qqbot/core/application/send/qqbot-send.service.ts @@ -287,7 +287,7 @@ export class QqbotSendService { retryable: false, sendLogId: log!.id, }); - await this.markFailedLog(log!.id, message); + await this.markFailedLog(log!.id, error.message); throw error; } await this.sendLogRepository.update( @@ -330,7 +330,7 @@ export class QqbotSendService { 'OneBot send failed', ); if (input.strict) { - const error = this.toStrictSendError(err, message, log!.id); + const error = this.toStrictSendError(err, log!.id); await this.markFailedLog(log!.id, error.message); throw error; } @@ -370,26 +370,21 @@ export class QqbotSendService { /** * Converts strict transport and infrastructure failures to stable delivery classifications. * @param err - The original failure. - * @param message - Its non-sensitive safe summary. * @param sendLogId - The pending log ID, if creation succeeded. * @returns A stable strict delivery error. */ - private toStrictSendError( - err: unknown, - message: string, - sendLogId: null | string, - ) { + private toStrictSendError(err: unknown, sendLogId: null | string) { if (err instanceof QqbotReverseWsActionError) { return new QqbotSendAttemptError({ code: err.code, - message, + message: err.message, retryable: true, sendLogId, }); } return new QqbotSendAttemptError({ code: 'onebot_disconnected', - message, + message: 'OneBot send failed', retryable: true, sendLogId, }); @@ -406,11 +401,11 @@ export class QqbotSendService { err: unknown, sendLogId: null | string, ): never { + if (strict) throw this.toStrictSendError(err, sendLogId); const message = this.toolsService.getErrorMessage( err, 'OneBot send failed', ); - if (strict) throw this.toStrictSendError(err, message, sendLogId); throwVbenError(message); throw new Error(message); } diff --git a/test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts b/test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts index 72bb9b6..853b9a9 100644 --- a/test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts +++ b/test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts @@ -873,11 +873,13 @@ describe('QqbotMessageSubscriptionService', () => { 'disable', async (service: QqbotMessageSubscriptionService) => service.setEnabled('100', false), + false, ], [ 'soft delete', async (service: QqbotMessageSubscriptionService) => service.remove('100'), + false, ], [ 'canonical config change', @@ -892,19 +894,26 @@ describe('QqbotMessageSubscriptionService', () => { }, sourceKey: SOURCE_KEY, }), + true, ], ])( - '%s commits config and cancels only exact claimable rows without rewriting snapshots', - async (_name, mutate) => { + '%s commits config and applies the exact identity-sensitive cancellation fence', + async (_name, mutate, cancelsProcessing) => { const harness = cancellationHarness(); const before = structuredClone(harness.deliveries); + const cancellableStatuses = [ + 'waiting_ddns', + 'pending', + 'retry', + ...(cancelsProcessing ? ['processing'] : []), + ]; await mutate(harness.service); expect(harness.deliveries).toEqual( before.map((delivery) => delivery.subscriptionId === '100' && - ['waiting_ddns', 'pending', 'retry'].includes(delivery.status) + cancellableStatuses.includes(delivery.status) ? { ...delivery, nextAttemptAt: null, @@ -916,7 +925,7 @@ describe('QqbotMessageSubscriptionService', () => { ); expect( (harness.deliveryUpdates[0].status as { _value: string[] })._value, - ).toEqual(['waiting_ddns', 'pending', 'retry']); + ).toEqual(cancellableStatuses); }, ); diff --git a/test/modules/qqbot/message-push/qqbot-strict-send.spec.ts b/test/modules/qqbot/message-push/qqbot-strict-send.spec.ts index 755c52b..8c1abde 100644 --- a/test/modules/qqbot/message-push/qqbot-strict-send.spec.ts +++ b/test/modules/qqbot/message-push/qqbot-strict-send.spec.ts @@ -228,7 +228,7 @@ describe('QQBot strict plain-text sender', () => { expect(sendLogRepository.update).toHaveBeenCalledWith( { id: 'log-1' }, expect.objectContaining({ - errorMessage: 'bad target', + errorMessage: 'OneBot rejected the send action', status: 'failed', }), ); @@ -291,6 +291,41 @@ describe('QQBot strict plain-text sender', () => { }); }); + it('redacts raw infrastructure details from both the typed error and failed send log', async () => { + const { busService, sendLogRepository, service } = createHarness(); + busService.publish.mockRejectedValue( + new Error( + 'password=broker-secret SELECT * FROM qqbot_private mqtt://admin@broker', + ), + ); + + await expect( + service.sendStrictPlainText({ + attemptNumber: 1, + deliveryId: 'delivery-1', + message: 'plain text', + selfId: '10001', + targetId: '20001', + targetType: 'group', + }), + ).rejects.toMatchObject({ + code: 'onebot_disconnected', + message: 'OneBot connection unavailable', + retryable: true, + sendLogId: 'log-1', + }); + expect(sendLogRepository.update).toHaveBeenCalledWith( + { id: 'log-1' }, + { + errorMessage: 'OneBot connection unavailable', + status: 'failed', + }, + ); + expect(JSON.stringify(sendLogRepository.update.mock.calls)).not.toContain( + 'broker-secret', + ); + }); + it('rejects an invalid strict target before rate limiting, logs, or transport', async () => { const { busService, @@ -366,6 +401,25 @@ describe('QQBot strict plain-text sender', () => { }); describe('QQBot reverse WS action classification', () => { + it.each([ + ['account_unavailable', 'Configured QQBot account is unavailable'], + ['invalid_target_type', 'Strict QQBot delivery target type is invalid'], + ['onebot_rejected', 'OneBot rejected the send action'], + ['onebot_timeout', 'OneBot send timed out'], + ['onebot_disconnected', 'OneBot connection unavailable'], + ['future_unknown_code', 'QQBot delivery failed'], + ])('maps strict-send code %s to a stable safe summary', (code, message) => { + const error = new QqbotSendAttemptError({ + code, + message: 'password=must-not-survive', + retryable: true, + sendLogId: null, + }); + + expect(error.message).toBe(message); + expect(error.message).not.toContain('must-not-survive'); + }); + it('keeps the typed strict send error fields explicit', () => { const error = new QqbotSendAttemptError({ code: 'onebot_timeout', diff --git a/test/modules/qqbot/message-push/system-message-delivery-runner.spec.ts b/test/modules/qqbot/message-push/system-message-delivery-runner.spec.ts index 58ebce0..1cd6b6b 100644 --- a/test/modules/qqbot/message-push/system-message-delivery-runner.spec.ts +++ b/test/modules/qqbot/message-push/system-message-delivery-runner.spec.ts @@ -279,6 +279,7 @@ function setup(seed: Partial = {}) { | null | ((authoritative: Store, where: Record) => void) = null; let beforePreparationDeliveryRead: null | ((draft: Store) => void) = null; + let afterClaimCommit: null | ((authoritative: Store) => void) = null; let pauseNextClaim: null | { entered: () => void; resume: Promise } = null; @@ -518,6 +519,11 @@ function setup(seed: Partial = {}) { repository(entity, draft, context, transactionLocks), }); mergeCommitted(draft, baseline); + if (context.claim && afterClaimCommit) { + const callback = afterClaimCommit; + afterClaimCommit = null; + callback(state); + } operations.push( context.claim ? 'claim:commit' @@ -543,6 +549,9 @@ function setup(seed: Partial = {}) { sender as never, ); return { + afterClaimCommit: (callback: null | ((authoritative: Store) => void)) => { + afterClaimCommit = callback; + }, adapter, beforeExternalUpdate: ( callback: @@ -800,6 +809,32 @@ describe('System message delivery runner direct preflight contracts', () => { }); }); + it('5 loses ownership when a canonical config change cancels the claimed old-identity row', async () => { + const harness = setup(); + harness.afterClaimCommit((state) => { + state.subscriptions[0].sourceConfig = { + ddnsRecordId: 'new-ddns-record', + portForwardId: RESOURCE_KEY, + }; + Object.assign(state.deliveries[0], { + nextAttemptAt: null, + processingLeaseUntil: null, + status: 'cancelled', + }); + }); + + await harness.runner.runOnce(NOW); + + expect(harness.adapter.resolveDelivery).not.toHaveBeenCalled(); + expect(harness.sender.sendStrictPlainText).not.toHaveBeenCalled(); + expect(harness.getState().deliveries[0]).toMatchObject({ + attemptCount: 1, + lastErrorCode: 'old_error', + processingLeaseUntil: null, + status: 'cancelled', + }); + }); + it('6-8 sends exact frozen input after both transactions commit and ignores current template identity', async () => { const harness = setup({ bindings: [binding({ templateId: 'new-template' })], @@ -1066,7 +1101,7 @@ describe('System message delivery runner direct preflight contracts', () => { expect(harness.getState().deliveries[0]).toMatchObject({ attemptCount: 1, lastErrorCode: 'onebot_timeout', - lastErrorMessage: 'OneBot action timeout', + lastErrorMessage: 'OneBot send timed out', nextAttemptAt: new KtDateTime(NOW.getTime() + 10_000), processingLeaseUntil: null, sendLogId: '90001', @@ -1074,6 +1109,31 @@ describe('System message delivery runner direct preflight contracts', () => { }); }); + it('17 never persists arbitrary text carried by an explicitly typed strict-send error', async () => { + const harness = setup(); + harness.sender.sendStrictPlainText.mockRejectedValueOnce( + new QqbotSendAttemptError({ + code: 'onebot_disconnected', + message: + 'password=delivery-secret SELECT * FROM qqbot_private connection=mysql://root@db', + retryable: true, + sendLogId: '90002', + }), + ); + + await harness.runner.runOnce(NOW); + + expect(harness.getState().deliveries[0]).toMatchObject({ + lastErrorCode: 'onebot_disconnected', + lastErrorMessage: 'OneBot connection unavailable', + sendLogId: '90002', + status: 'retry', + }); + expect(JSON.stringify(harness.getState().deliveries[0])).not.toContain( + 'delivery-secret', + ); + }); + it('18 caps exponential backoff at fifteen minutes', async () => { const harness = setup({ deliveries: [delivery({ attemptCount: 19 })], @@ -1232,6 +1292,65 @@ describe('System message delivery runner direct preflight contracts', () => { ); expect(harness.getState().deliveries[0].processingLeaseUntil).toBeNull(); }); + + it('samples a fresh production clock for each claim after a slow prior send', async () => { + jest.useFakeTimers().setSystemTime(NOW); + try { + const harness = setup({ + deliveries: [ + delivery({ id: '801', publishTargetId: '701' }), + delivery({ id: '802', publishTargetId: '702' }), + ], + targets: [target(), target({ id: '702' })], + }); + const observedLeases: Array = []; + harness.beforeExternalUpdate((state) => { + const processing = state.deliveries.find( + (item) => item.status === 'processing', + ); + observedLeases.push(processing?.processingLeaseUntil ?? null); + }); + harness.sender.sendStrictPlainText.mockImplementation(async () => { + if (harness.sender.sendStrictPlainText.mock.calls.length === 1) { + jest.setSystemTime(NOW.getTime() + 31_000); + } + return { logId: 'send-log' }; + }); + + await harness.runner.runOnce(); + + expect(observedLeases).toEqual([ + new KtDateTime(NOW.getTime() + SYSTEM_MESSAGE_LEASE_MS), + new KtDateTime(NOW.getTime() + 31_000 + SYSTEM_MESSAGE_LEASE_MS), + ]); + } finally { + jest.useRealTimers(); + } + }); + + it('schedules a production retry from the fresh post-I/O clock', async () => { + jest.useFakeTimers().setSystemTime(NOW); + try { + const harness = setup(); + harness.sender.sendStrictPlainText.mockImplementationOnce(async () => { + jest.setSystemTime(NOW.getTime() + 31_000); + throw new QqbotSendAttemptError({ + code: 'onebot_timeout', + message: 'raw timeout detail', + retryable: true, + sendLogId: 'slow-log', + }); + }); + + await harness.runner.runOnce(); + + expect(harness.getState().deliveries[0].nextAttemptAt).toEqual( + new KtDateTime(NOW.getTime() + 41_000), + ); + } finally { + jest.useRealTimers(); + } + }); }); describe('System message coordinator direct lifecycle contracts', () => {