From 97a560820f0d82f458e5be7a32805534fddff411 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 27 Sep 2026 16:38:46 +0400 Subject: [PATCH] feat(core): queue runner with retries, stale-sent recovery and re-entrancy guard Refs #1 Co-Authored-By: Claude Opus 5.5 --- src/core/queue.ts | 123 +++++++++++++++++++++++ tests/queue.test.ts | 240 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 363 insertions(+) create mode 100644 src/core/queue.ts create mode 100644 tests/queue.test.ts diff --git a/src/core/queue.ts b/src/core/queue.ts new file mode 100644 index 0000000..55f2021 --- /dev/null +++ b/src/core/queue.ts @@ -0,0 +1,123 @@ +import { parseRecord, readVoiceFields, renderNote } from './note'; +import type { Env, NoteStore, RecordingStore, TranscribeResult, Transcriber } from './ports'; +import { NO_SPEECH, transition, type VoiceRecord } from './record'; +import { due, isStaleSent, staleOwn, type Located } from './schedule'; + +export interface QueueDeps { + recordings: RecordingStore; + notes: NoteStore; + transcriber: Transcriber; + env: Env; +} + +export const INTERRUPTED_SEND = 'отправка прервана'; + +interface Change { + record: VoiceRecord; + text?: string; +} + +export class QueueRunner { + private running = false; + private rerun = false; + private readonly controller = new AbortController(); + + constructor(private readonly deps: QueueDeps) {} + + // Повторный вызов во время работы не запускает параллельный проход, + // а просит ровно один дополнительный после текущего. + async tick(): Promise { + if (this.running) { + this.rerun = true; + return; + } + this.running = true; + try { + do { + this.rerun = false; + await this.pass(); + } while (this.rerun); + } finally { + this.running = false; + } + } + + dispose(): void { + this.controller.abort(); + } + + private async pass(): Promise { + const { env } = this.deps; + const items = await this.load(); + const now = env.now(); + for (const item of staleOwn(items, now, env.deviceId)) { + await this.update( + item.path, + (r) => r.id === item.record.id && r.device === env.deviceId && isStaleSent(r, now), + (r) => ({ record: transition(r, { type: 'temporary', message: INTERRUPTED_SEND }, now) }), + ); + } + for (const item of due(items, now, env.deviceId)) await this.send(item); + } + + private async send(item: Located): Promise { + const { recordings, transcriber, env } = this.deps; + const now = env.now(); + // Список мог устареть (metadataCache) — решение принимаем по актуальному файлу. + const sent = await this.update( + item.path, + (r) => r.id === item.record.id && due([{ path: item.path, record: r }], now, env.deviceId).length > 0, + (r) => ({ record: transition(r, { type: 'send' }, now) }), + ); + if (!sent) return; + + let result: TranscribeResult; + try { + const audio = await recordings.read(sent.audio); + result = await transcriber.transcribe(audio, sent.mime, this.controller.signal); + } catch (e) { + result = { ok: false, kind: 'temporary', message: e instanceof Error ? e.message : String(e) }; + } + if (result.ok && result.text.trim() === '') { + result = { ok: false, kind: 'permanent', message: NO_SPEECH }; + } + + const done = env.now(); + const final = result; + await this.update( + item.path, + (r) => r.id === sent.id && r.status === 'sent', + (r) => + final.ok + ? { record: transition(r, { type: 'success' }, done), text: final.text } + : { record: transition(r, { type: final.kind, message: final.message }, done) }, + ); + } + + // Атомарно: перечитать запись из актуального содержимого, проверить ожидание, + // применить изменение. null — заметки нет или запись не в ожидаемом состоянии. + private async update( + path: string, + expect: (r: VoiceRecord) => boolean, + change: (r: VoiceRecord) => Change, + ): Promise { + const box: { record: VoiceRecord | null } = { record: null }; + const exists = await this.deps.notes.process(path, (content) => { + const current = parseRecord(readVoiceFields(content)); + if (!current || !expect(current)) return content; + const next = change(current); + box.record = next.record; + return renderNote(content, next.record, next.text); + }); + return exists ? box.record : null; + } + + private async load(): Promise { + const located: Located[] = []; + for (const { path, frontmatter } of await this.deps.notes.list()) { + const record = parseRecord(frontmatter); + if (record) located.push({ path, record }); + } + return located; + } +} diff --git a/tests/queue.test.ts b/tests/queue.test.ts new file mode 100644 index 0000000..0abae77 --- /dev/null +++ b/tests/queue.test.ts @@ -0,0 +1,240 @@ +import { describe, expect, it } from 'vitest'; +import { initialNote, readVoiceFields } from '../src/core/note'; +import { QueueRunner } from '../src/core/queue'; +import type { VoiceRecord } from '../src/core/record'; +import { MemoryNotes, MemoryRecordings, ScriptedTranscriber, TestEnv, type Step } from './fakes'; + +const T0 = new Date('2026-09-27T11:15:03.000Z'); +const ok = (text: string): Step => ({ ok: true, text }); +const temporary = (message: string): Step => ({ ok: false, kind: 'temporary', message }); +const permanent = (message: string): Step => ({ ok: false, kind: 'permanent', message }); + +function setup(steps: Step[] = []) { + const env = new TestEnv(T0); + const notes = new MemoryNotes(); + const recordings = new MemoryRecordings(); + const transcriber = new ScriptedTranscriber(steps); + const runner = new QueueRunner({ recordings, notes, transcriber, env }); + return { env, notes, recordings, transcriber, runner }; +} +type Ctx = ReturnType; + +function rec(id: string, over: Partial = {}): VoiceRecord { + return { id, status: 'saved', device: 'test-device', audio: `voice/source/${id}.m4a`, mime: 'audio/mp4', attempts: 0, ...over }; +} + +function seed(ctx: Ctx, record: VoiceRecord, path = `voice/inbox/${record.id}.md`): string { + ctx.notes.files.set(path, initialNote(record, 0)); + ctx.recordings.files.set(record.audio, new Uint8Array([1, 2, 3]).buffer); + return path; +} + +const fields = (ctx: Ctx, path: string) => readVoiceFields(ctx.notes.files.get(path)!); + +describe('QueueRunner.tick', () => { + it('sends a saved record and inserts the trimmed transcript', async () => { + const ctx = setup([ok(' привет ')]); + const path = seed(ctx, rec('a')); + await ctx.runner.tick(); + + expect(ctx.transcriber.calls).toHaveLength(1); + expect(new Uint8Array(ctx.transcriber.calls[0].audio)).toEqual(new Uint8Array([1, 2, 3])); + expect(ctx.transcriber.calls[0].mime).toBe('audio/mp4'); + expect(fields(ctx, path)['voice-status']).toBe('inserted'); + expect(fields(ctx, path)['voice-sent-at']).toBeUndefined(); + expect(ctx.notes.files.get(path)).toContain('\n\nпривет\n'); + expect(ctx.notes.files.get(path)).not.toContain('%%voice-note'); + }); + + it('network lost, then back: retry at +30s, then inserted', async () => { + const ctx = setup([temporary('offline'), ok('текст')]); + const path = seed(ctx, rec('a')); + + await ctx.runner.tick(); + expect(fields(ctx, path)).toMatchObject({ + 'voice-status': 'retry', + 'voice-attempts': '1', + 'voice-next-attempt': '2026-09-27T11:15:33.000Z', + 'voice-error': 'offline', + }); + expect(ctx.notes.files.get(path)).toContain('Не удалось отправить, повтор в 14:15 (попытка 1/5)'); + + ctx.env.advance(17_000); // 11:15:20 — рано + await ctx.runner.tick(); + expect(ctx.transcriber.calls).toHaveLength(1); + + ctx.env.advance(13_000); // 11:15:33 — пора + await ctx.runner.tick(); + expect(ctx.transcriber.calls).toHaveLength(2); + expect(fields(ctx, path)['voice-status']).toBe('inserted'); + }); + + it('five temporary failures → failed with the last message', async () => { + const ctx = setup([temporary('e1'), temporary('e2'), temporary('e3'), temporary('e4'), temporary('e5')]); + const path = seed(ctx, rec('a')); + for (let i = 0; i < 5; i++) { + await ctx.runner.tick(); + ctx.env.advance(3_600_000); + } + await ctx.runner.tick(); + expect(ctx.transcriber.calls).toHaveLength(5); + expect(fields(ctx, path)).toMatchObject({ 'voice-status': 'failed', 'voice-attempts': '5', 'voice-error': 'e5' }); + }); + + it('permanent error → failed at once', async () => { + const ctx = setup([permanent('401 Unauthorized')]); + const path = seed(ctx, rec('a')); + await ctx.runner.tick(); + expect(fields(ctx, path)).toMatchObject({ 'voice-status': 'failed', 'voice-attempts': '1' }); + expect(ctx.notes.files.get(path)).toContain('Не распознано: 401 Unauthorized'); + }); + + it('whitespace-only transcript → failed no-speech', async () => { + const ctx = setup([ok(' \n\t ')]); + const path = seed(ctx, rec('a')); + await ctx.runner.tick(); + expect(fields(ctx, path)).toMatchObject({ 'voice-status': 'failed', 'voice-error': 'no-speech' }); + expect(ctx.notes.files.get(path)).toContain('> [!info] В записи не распознано речи'); + }); + + it('exception from transcriber is a temporary failure', async () => { + const ctx = setup([new Error('socket hang up')]); + const path = seed(ctx, rec('a')); + await ctx.runner.tick(); + expect(fields(ctx, path)).toMatchObject({ 'voice-status': 'retry', 'voice-error': 'socket hang up' }); + }); + + it('missing audio file is a temporary failure', async () => { + const ctx = setup([]); + const path = seed(ctx, rec('a')); + ctx.recordings.files.clear(); + await ctx.runner.tick(); + expect(ctx.transcriber.calls).toHaveLength(0); + expect(fields(ctx, path)['voice-status']).toBe('retry'); + }); + + it("does not send other devices' records", async () => { + const ctx = setup([]); + const path = seed(ctx, rec('a', { device: 'iphone' })); + await ctx.runner.tick(); + expect(ctx.transcriber.calls).toHaveLength(0); + expect(fields(ctx, path)['voice-status']).toBe('saved'); + }); + + it('own sent older than 10 min → retry without a request; younger untouched', async () => { + const ctx = setup([]); + const stale = seed(ctx, rec('a', { status: 'sent', sentAt: '2026-09-27T11:04:00.000Z' })); + const fresh = seed(ctx, rec('b', { status: 'sent', sentAt: '2026-09-27T11:13:03.000Z' })); + const before = ctx.notes.files.get(fresh); + await ctx.runner.tick(); + expect(ctx.transcriber.calls).toHaveLength(0); + expect(fields(ctx, stale)).toMatchObject({ 'voice-status': 'retry', 'voice-attempts': '1', 'voice-error': 'отправка прервана' }); + expect(ctx.notes.files.get(fresh)).toBe(before); + }); + + it('processes records one at a time, oldest first', async () => { + let inFlight = 0; + let maxInFlight = 0; + const order: string[] = []; + const slow: Step = async (audio) => { + inFlight++; + maxInFlight = Math.max(maxInFlight, inFlight); + order.push(String(new Uint8Array(audio)[0])); + await new Promise((resolve) => setTimeout(resolve, 5)); + inFlight--; + return { ok: true, text: 't' }; + }; + const ctx = setup([slow, slow, slow]); + for (const [id, byte] of [['c', 3], ['a', 1], ['b', 2]] as const) { + seed(ctx, rec(id)); + ctx.recordings.files.set(`voice/source/${id}.m4a`, new Uint8Array([byte]).buffer); + } + await ctx.runner.tick(); + expect(maxInFlight).toBe(1); + expect(order).toEqual(['1', '2', '3']); + }); + + it('tick during tick runs exactly one extra pass', async () => { + const ctx: Ctx = setup(); + ctx.transcriber.steps.push(async () => { + void ctx.runner.tick(); + void ctx.runner.tick(); + void ctx.runner.tick(); + return { ok: true, text: 't' }; + }); + seed(ctx, rec('a')); + await ctx.runner.tick(); + expect(ctx.notes.listCalls).toBe(2); + expect(ctx.transcriber.calls).toHaveLength(1); + }); + + it('note deleted during the request: result dropped, next record processed', async () => { + const ctx: Ctx = setup(); + const first = seed(ctx, rec('a')); + const second = seed(ctx, rec('b')); + ctx.transcriber.steps.push(async () => { + ctx.notes.files.delete(first); + return { ok: true, text: 'один' }; + }, ok('два')); + await ctx.runner.tick(); + expect(ctx.notes.files.has(first)).toBe(false); + expect(fields(ctx, second)['voice-status']).toBe('inserted'); + }); + + it('note deleted before send: skipped without a request', async () => { + const ctx = setup([ok('два')]); + const second = seed(ctx, rec('b')); + ctx.notes.staleList = [ + { path: 'voice/inbox/gone.md', frontmatter: readVoiceFields(initialNote(rec('a'), 0)) }, + { path: second, frontmatter: fields(ctx, second) }, + ]; + await ctx.runner.tick(); + expect(ctx.transcriber.calls).toHaveLength(1); + expect(fields(ctx, second)['voice-status']).toBe('inserted'); + }); + + it('stale list says saved, file says sent: no second request', async () => { + const ctx = setup([]); + const path = seed(ctx, rec('a', { status: 'sent', sentAt: '2026-09-27T11:15:00.000Z' })); + ctx.notes.staleList = [{ path, frontmatter: readVoiceFields(initialNote(rec('a'), 0)) }]; + const before = ctx.notes.files.get(path); + await ctx.runner.tick(); + expect(ctx.transcriber.calls).toHaveLength(0); + expect(ctx.notes.files.get(path)).toBe(before); + }); + + it('stale list says saved, file says inserted: untouched', async () => { + const ctx = setup([ok('x')]); + const path = seed(ctx, rec('a')); + await ctx.runner.tick(); + const done = ctx.notes.files.get(path); + ctx.notes.staleList = [{ path, frontmatter: readVoiceFields(initialNote(rec('a'), 0)) }]; + await ctx.runner.tick(); + expect(ctx.transcriber.calls).toHaveLength(1); + expect(ctx.notes.files.get(path)).toBe(done); + }); + + it('finds records outside voice/inbox', async () => { + const ctx = setup([ok('x')]); + const path = seed(ctx, rec('a'), 'projects/перенесённая.md'); + await ctx.runner.tick(); + expect(fields(ctx, path)['voice-status']).toBe('inserted'); + }); + + it('ignores notes with a broken status', async () => { + const ctx = setup([]); + const path = seed(ctx, rec('a')); + ctx.notes.files.set(path, ctx.notes.files.get(path)!.replace('"saved"', '"foo"')); + await expect(ctx.runner.tick()).resolves.toBeUndefined(); + expect(ctx.transcriber.calls).toHaveLength(0); + }); + + it('dispose aborts the signal given to the transcriber', async () => { + const ctx = setup([ok('x')]); + seed(ctx, rec('a')); + await ctx.runner.tick(); + expect(ctx.transcriber.calls[0].signal.aborted).toBe(false); + ctx.runner.dispose(); + expect(ctx.transcriber.calls[0].signal.aborted).toBe(true); + }); +});