feat(core): queue runner with retries, stale-sent recovery and re-entrancy guard
Refs #1 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
b126e4bf8b
commit
97a560820f
123
src/core/queue.ts
Normal file
123
src/core/queue.ts
Normal file
@ -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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<VoiceRecord | null> {
|
||||
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<Located[]> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
240
tests/queue.test.ts
Normal file
240
tests/queue.test.ts
Normal file
@ -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<typeof setup>;
|
||||
|
||||
function rec(id: string, over: Partial<VoiceRecord> = {}): 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);
|
||||
});
|
||||
});
|
||||
Loading…
x
Reference in New Issue
Block a user