fix(core): stop sending after QueueRunner.dispose()

Refs #1

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-09-27 16:42:57 +04:00
parent d7ec88a91e
commit 626d832bd1
2 changed files with 29 additions and 2 deletions

View File

@ -27,6 +27,7 @@ export class QueueRunner {
// Повторный вызов во время работы не запускает параллельный проход, // Повторный вызов во время работы не запускает параллельный проход,
// а просит ровно один дополнительный после текущего. // а просит ровно один дополнительный после текущего.
async tick(): Promise<void> { async tick(): Promise<void> {
if (this.controller.signal.aborted) return;
if (this.running) { if (this.running) {
this.rerun = true; this.rerun = true;
return; return;
@ -36,7 +37,7 @@ export class QueueRunner {
do { do {
this.rerun = false; this.rerun = false;
await this.pass(); await this.pass();
} while (this.rerun); } while (this.rerun && !this.controller.signal.aborted);
} finally { } finally {
this.running = false; this.running = false;
} }
@ -57,10 +58,14 @@ export class QueueRunner {
(r) => ({ record: transition(r, { type: 'temporary', message: INTERRUPTED_SEND }, now) }), (r) => ({ record: transition(r, { type: 'temporary', message: INTERRUPTED_SEND }, now) }),
); );
} }
for (const item of due(items, now, env.deviceId)) await this.send(item); for (const item of due(items, now, env.deviceId)) {
if (this.controller.signal.aborted) break;
await this.send(item);
}
} }
private async send(item: Located): Promise<void> { private async send(item: Located): Promise<void> {
if (this.controller.signal.aborted) return;
const { recordings, transcriber, env } = this.deps; const { recordings, transcriber, env } = this.deps;
const now = env.now(); const now = env.now();
// Список мог устареть (metadataCache) — решение принимаем по актуальному файлу. // Список мог устареть (metadataCache) — решение принимаем по актуальному файлу.

View File

@ -237,4 +237,26 @@ describe('QueueRunner.tick', () => {
ctx.runner.dispose(); ctx.runner.dispose();
expect(ctx.transcriber.calls[0].signal.aborted).toBe(true); expect(ctx.transcriber.calls[0].signal.aborted).toBe(true);
}); });
it('dispose during a request stops the pass without touching later due records', async () => {
const ctx = setup();
const first = seed(ctx, rec('a'));
const second = seed(ctx, rec('b'));
const beforeSecond = ctx.notes.files.get(second);
ctx.transcriber.steps.push(async () => {
ctx.runner.dispose();
throw new Error('aborted');
});
await ctx.runner.tick();
expect(ctx.transcriber.calls).toHaveLength(1);
expect(fields(ctx, first)).toMatchObject({ 'voice-status': 'retry', 'voice-attempts': '1' });
expect(fields(ctx, second)['voice-status']).toBe('saved');
expect(ctx.notes.files.get(second)).toBe(beforeSecond);
const listCallsBefore = ctx.notes.listCalls;
await ctx.runner.tick();
expect(ctx.notes.listCalls).toBe(listCallsBefore);
});
}); });