diff --git a/api/src/protocol-logs/protocol-logs.service.spec.ts b/api/src/protocol-logs/protocol-logs.service.spec.ts index 6fba96b..eec40ee 100644 --- a/api/src/protocol-logs/protocol-logs.service.spec.ts +++ b/api/src/protocol-logs/protocol-logs.service.spec.ts @@ -81,13 +81,53 @@ describe('ProtocolLogsService', () => { const service = new ProtocolLogsService(prisma as never); await service.list({ createdAtFrom: '2026-08-21', createdAtTo: '2026-08-27' }); - expect(prisma.protocolInteractionLog.findMany).toHaveBeenCalledWith(expect.objectContaining({ - where: expect.objectContaining({ - createdAt: { - gte: new Date('2026-08-20T16:00:00.000Z'), - lte: new Date('2026-08-27T15:59:59.999Z'), - }, + expect(prisma.protocolInteractionLog.findMany).toHaveBeenCalledWith( + expect.objectContaining({ + where: expect.objectContaining({ + createdAt: { + gte: new Date('2026-08-20T16:00:00.000Z'), + lte: new Date('2026-08-27T15:59:59.999Z'), + }, + }), }), - })); + ); + }); + + it('waits for an active flush and persists every buffered batch before reporting success', async () => { + const previousBatchSize = process.env.PROTOCOL_LOG_BATCH_SIZE; + process.env.PROTOCOL_LOG_BATCH_SIZE = '2'; + let releaseFirstBatch!: (value: { count: number }) => void; + prisma.protocolInteractionLog.createMany + .mockImplementationOnce( + () => + new Promise((resolve) => { + releaseFirstBatch = resolve; + }), + ) + .mockResolvedValue({ count: 1 }); + + try { + const service = new ProtocolLogsService(prisma as never); + service.recordMany( + [1, 2, 3].map((sequenceId) => ({ + protocol: 'cmpp' as const, + direction: 'channel_to_platform' as const, + eventType: 'deliver_receipt', + status: 'success' as const, + detail: { sequenceId }, + })), + ); + + const completed = service.flushNow(); + expect(prisma.protocolInteractionLog.createMany).toHaveBeenCalledTimes(1); + releaseFirstBatch({ count: 2 }); + + await expect(completed).resolves.toBe(true); + expect(prisma.protocolInteractionLog.createMany).toHaveBeenCalledTimes(2); + expect(prisma.protocolInteractionLog.createMany.mock.calls.flatMap(([input]) => input.data)).toHaveLength(3); + } finally { + if (previousBatchSize === undefined) delete process.env.PROTOCOL_LOG_BATCH_SIZE; + else process.env.PROTOCOL_LOG_BATCH_SIZE = previousBatchSize; + } }); }); diff --git a/api/src/protocol-logs/protocol-logs.service.ts b/api/src/protocol-logs/protocol-logs.service.ts index c0dac3f..09f43d0 100644 --- a/api/src/protocol-logs/protocol-logs.service.ts +++ b/api/src/protocol-logs/protocol-logs.service.ts @@ -47,7 +47,7 @@ export class ProtocolLogsService implements OnModuleInit, OnModuleDestroy { private readonly buffer: Prisma.ProtocolInteractionLogCreateManyInput[] = []; private flushTimer?: ReturnType; private retentionTimer?: ReturnType; - private flushing = false; + private flushPromise?: Promise; constructor(private readonly prisma: PrismaService) {} @@ -142,22 +142,32 @@ export class ProtocolLogsService implements OnModuleInit, OnModuleDestroy { return { items, total, page, pageSize, eventTypes: eventTypes.map((item) => item.eventType) }; } - private async flush(): Promise { - if (this.flushing) return false; - if (this.buffer.length === 0) return true; - this.flushing = true; - const batch = this.buffer.splice(0, positiveEnv('PROTOCOL_LOG_BATCH_SIZE', 100)); - try { - await this.prisma.protocolInteractionLog.createMany({ data: batch }); - } catch (error) { - this.buffer.unshift(...batch); - this.logger.error(`Protocol log batch write failed (${batch.length} events)`, error instanceof Error ? error.stack : String(error)); - return false; - } finally { - this.flushing = false; - if (this.buffer.length > 0) setImmediate(() => void this.flush()); + private flush(): Promise { + if (this.flushPromise) return this.flushPromise; + if (this.buffer.length === 0) return Promise.resolve(true); + + const flushPromise = this.flushBufferedBatches().finally(() => { + if (this.flushPromise === flushPromise) this.flushPromise = undefined; + }); + this.flushPromise = flushPromise; + return flushPromise; + } + + private async flushBufferedBatches(): Promise { + while (this.buffer.length > 0) { + const batch = this.buffer.splice(0, positiveEnv('PROTOCOL_LOG_BATCH_SIZE', 100)); + try { + await this.prisma.protocolInteractionLog.createMany({ data: batch }); + } catch (error) { + this.buffer.unshift(...batch); + this.logger.error( + `Protocol log batch write failed (${batch.length} events)`, + error instanceof Error ? error.stack : String(error), + ); + return false; + } } - return true; + return true; } private async purgeExpired() { diff --git a/docs/system-functional-test-cases.md b/docs/system-functional-test-cases.md index 8643ef7..dedbcb4 100644 --- a/docs/system-functional-test-cases.md +++ b/docs/system-functional-test-cases.md @@ -3884,6 +3884,7 @@ npm run verify:phase8 - `TC-DELIVERY-AUTO-014`:分别配置仅CMPP、仅HTTP、CMPP+HTTP、两者均关闭四种应用状态;回执与上行分别只产生CMPP下游记录、HTTP Webhook事件、两者各一条、均不产生。修改历史手工投递模式不得改变自动计算结果。 - `TC-HTTP-WEBHOOK-015`:运营端关闭HTTP接口后,回执和上行Webhook地址输入框仍显示且可保存;任一地址保存为空时删除对应有效端点,后续不推送该类HTTP事件,另一非空地址不受影响。 - `TC-PROTOCOL-LOG-016`:在线企业应用收到回执或上行 `CMPP_DELIVER` 并返回 `CMPP_DELIVER_RESP`;通讯日志各出现一条“平台→企业应用/DELIVER”和“企业应用→平台/DELIVER_RESP”,结果、消息号、序列号和投递记录一致,下游投递记录仍独立展示发送、ACK和重试状态。 +- `TC-PROTOCOL-LOG-017`:独立协议日志Worker一次读取数量大于单次数据库写批次,且首批写库尚未完成时再次请求flush;Worker必须等待同一flush完整写完全部批次后才返回成功并ACK/XDEL对应Stream事件,处理中不得因30秒自动认领重复落库,写库失败时保留pending供重试且不得提前ACK。 ## 2026-07-26 企业应用停用与回执清算专项 - `APP-DISABLE-001`:应用无待清算数据时点击停用,直接进入已停用并断开该账号全部CMPP连接。