obsidian-voice-note/tests/queue.test.ts
Andrey Epifancev 972976b16a fix(core): claimAll reclaims own work; isolate per-record queue errors
- I-1 (F-1): claimAll() now also moves this device's own `retry` records
  and own stale `sent` records to `saved` via the `claim` transition
  (device = own), same as it already does for other devices' claimable
  records, then runs a normal tick() so everything sends without waiting
  for the next retry time. The returned count still only reflects records
  reassigned from other devices. Adds `ownReclaimable()` to schedule.ts.
- M-1 (F-2): pass() wraps each per-record operation (staleOwn update and
  each send) in try/catch, remembers the first error, keeps processing the
  rest of the due queue, and rethrows the first error after both loops
  complete — so one permanently failing record no longer blocks newer
  records behind it. list()/load() failures still reject tick()
  immediately, as before.
- F-4a: send()'s post-request update now also requires
  `r.device === env.deviceId && r.sentAt === sent.sentAt`, so a late
  transcription result can never overwrite a record another device (or a
  newer send) has since taken over.
- F-4b: pass()'s staleOwn loop now also stops when the abort signal fires;
  retry() and claimAll() return early (false / 0) without writing once
  disposed.
- F-4d: retry() no-op it.each now also covers `sent` (own, fresh).

Refs #1

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-27 17:12:31 +04:00

312 lines
12 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import { describe, expect, it } from 'vitest';
import { initialNote, patchFrontmatter, 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);
});
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);
});
it('one record throwing does not block the rest; tick rejects after the pass completes', async () => {
const ctx = setup([ok('два')]);
const first = seed(ctx, rec('a'));
const second = seed(ctx, rec('b'));
const boom = new Error('boom');
const originalProcess = ctx.notes.process.bind(ctx.notes);
ctx.notes.process = async (path, fn) => {
if (path === first) throw boom;
return originalProcess(path, fn);
};
await expect(ctx.runner.tick()).rejects.toBe(boom);
expect(ctx.transcriber.calls).toHaveLength(1);
expect(fields(ctx, second)['voice-status']).toBe('inserted');
});
it('post-request guard: result is dropped if another device already re-sent the record', async () => {
const ctx = setup([]);
const path = seed(ctx, rec('a'));
ctx.transcriber.steps.push(async () => {
const current = ctx.notes.files.get(path)!;
ctx.notes.files.set(
path,
patchFrontmatter(current, { 'voice-device': '"iphone"', 'voice-sent-at': '"2026-09-27T11:16:00.000Z"' }),
);
return { ok: true, text: 'x' };
});
await ctx.runner.tick();
expect(fields(ctx, path)).toMatchObject({
'voice-status': 'sent',
'voice-device': 'iphone',
'voice-sent-at': '2026-09-27T11:16:00.000Z',
});
});
it('sent status and sentAt are on disk while the request is in flight', async () => {
const ctx = setup();
const path = seed(ctx, rec('a'));
ctx.transcriber.steps.push(async () => {
expect(fields(ctx, path)).toMatchObject({ 'voice-status': 'sent', 'voice-sent-at': T0.toISOString() });
return { ok: true, text: 'x' };
});
await ctx.runner.tick();
expect(fields(ctx, path)['voice-status']).toBe('inserted');
});
});