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

fix: 收紧系统消息投递边界

Browse Source
This commit is contained in:
sunlei 2026-07-24 13:32:11 +08:00
parent d2fd59dcd4
commit d74545d17b
8 changed files with 288 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/modules/qqbot/core/application/account/qqbot-account.service.ts
View File

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

26
src/modules/qqbot/core/application/message-push/qqbot-message-subscription.service.ts
View File

@ -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, {
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<QqbotMessageDelivery, 'subscriptionId'>,
includeProcessing = false,
): Promise<void> {
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,

75
src/modules/qqbot/core/application/message-push/system-message-delivery-runner.service.ts
View File

@ -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<number> {
async runOnce(now?: Date): Promise<number> {
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<void> {
if (now.getTime() >= token.delivery.expiresAt.getTime()) {
private async processClaim(
token: ClaimToken,
fixedNow?: Date,
): Promise<void> {
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<PreparedDelivery> {
private async prepare(token: ClaimToken): Promise<PreparedDelivery> {
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);
}
}

19
src/modules/qqbot/core/application/send/qqbot-send.error.ts
View File

@ -1,5 +1,18 @@
import type { QqbotSendAttemptErrorOptions } from '../../contract/message-push/qqbot-message-push.types';
const STRICT_SEND_ERROR_SUMMARIES: Readonly<Record<string, string>> = {
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;

17
src/modules/qqbot/core/application/send/qqbot-send.service.ts
View File

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

17
test/modules/qqbot/message-push/qqbot-message-subscription.service.spec.ts
View File

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

56
test/modules/qqbot/message-push/qqbot-strict-send.spec.ts
View File

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

121
test/modules/qqbot/message-push/system-message-delivery-runner.spec.ts
View File

@ -279,6 +279,7 @@ function setup(seed: Partial<Store> = {}) {
| null
| ((authoritative: Store, where: Record<string, unknown>) => void) = null;
let beforePreparationDeliveryRead: null | ((draft: Store) => void) = null;
let afterClaimCommit: null | ((authoritative: Store) => void) = null;
let pauseNextClaim: null | { entered: () => void; resume: Promise<void> } =
null;
@ -518,6 +519,11 @@ function setup(seed: Partial<Store> = {}) {
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<Store> = {}) {
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<Date | null> = [];
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', () => {