aep/obsidian-voice-note
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: core recording lifecycle (#1) #2

Merged
aep merged 22 commits from feat/1-core-recording-lifecycle into main 2026-09-27 18:24:26 +03:00
Conversation 0 Commits 22 Files Changed 36 +138 -17
4 changed files with 138 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 972976b16a - Show all commits

60
src/core/queue.ts
View File

@ -1,7 +1,7 @@
import { parseRecord, readVoiceFields, renderNote } from './note';
import type { Env, NoteStore, RecordingStore, TranscribeResult, Transcriber } from './ports';
import { NO_SPEECH, transition, type VoiceRecord } from './record';
import { claimable, due, isStaleSent, staleOwn, type Located } from './schedule';
import { claimable, due, isStaleSent, ownReclaimable, staleOwn, type Located } from './schedule';
export interface QueueDeps {
recordings: RecordingStore;
@ -26,6 +26,9 @@ export class QueueRunner {
// Повторный вызов во время работы не запускает параллельный проход,
// а просит ровно один дополнительный после текущего.
// Отклоняется (rejects), если list() бросил исключение, или если обработка
// какой-то записи бросила исключение — в этом случае остальной проход всё
// равно завершается, и первая ошибка пробрасывается уже после него.
async tick(): Promise<void> {
if (this.controller.signal.aborted) return;
if (this.running) {
@ -50,6 +53,7 @@ export class QueueRunner {
// «Повторить распознавание»: failed/retry → saved, счётчик с нуля, запись
// переходит текущему устройству (иначе чужую запись здесь не отправить).
async retry(path: string): Promise<boolean> {
if (this.controller.signal.aborted) return false;
const { env } = this.deps;
const changed = await this.update(
path,
@ -63,11 +67,17 @@ export class QueueRunner {
// «Распознать все ожидающие здесь»: чужие saved/retry/зависшие sent → своё
// устройство, статус saved (без ожидания повтора), счётчик сохраняется.
// Дополнительно так же разбираются свои retry (независимо от
// voice-next-attempt) и свои зависшие sent — тоже без ожидания. После
// этого — обычный tick(). Возвращаемое число — только записи, переданные с
// других устройств; свои не считаются.
async claimAll(): Promise<number> {
if (this.controller.signal.aborted) return 0;
const { env } = this.deps;
const now = env.now();
const items = await this.load();
let claimed = 0;
for (const item of claimable(await this.load(), now, env.deviceId)) {
for (const item of claimable(items, now, env.deviceId)) {
const changed = await this.update(
item.path,
(r) => r.id === item.record.id && claimable([{ path: item.path, record: r }], now, env.deviceId).length > 0,
@ -75,25 +85,54 @@ export class QueueRunner {
);
if (changed) claimed++;
}
for (const item of ownReclaimable(items, now, env.deviceId)) {
await this.update(
item.path,
(r) =>
r.id === item.record.id && ownReclaimable([{ path: item.path, record: r }], now, env.deviceId).length > 0,
(r) => ({ record: transition(r, { type: 'claim', device: env.deviceId }, now) }),
);
}
await this.tick();
return claimed;
}
// Ошибка одной записи (process()/read() бросили исключение) не блокирует
// остальные: она запоминается, проход продолжается, и первая такая ошибка
// пробрасывается только после того, как обе очереди пройдены целиком.
private async pass(): Promise<void> {
const { env } = this.deps;
const items = await this.load();
const now = env.now();
let hasError = false;
let firstError: unknown;
const remember = (e: unknown) => {
if (!hasError) {
hasError = true;
firstError = e;
}
};
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) }),
);
if (this.controller.signal.aborted) break;
try {
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) }),
);
} catch (e) {
remember(e);
}
}
for (const item of due(items, now, env.deviceId)) {
if (this.controller.signal.aborted) break;
await this.send(item);
try {
await this.send(item);
} catch (e) {
remember(e);
}
}
if (hasError) throw firstError;
}
private async send(item: Located): Promise<void> {
@ -121,9 +160,12 @@ export class QueueRunner {
const done = env.now();
const final = result;
// Пока запрос был в полёте, запись могла уйти к другому устройству (или
// отправиться заново) — проверяем, что это всё ещё именно наша отправка,
// прежде чем записывать результат поверх неё.
await this.update(
item.path,
(r) => r.id === sent.id && r.status === 'sent',
(r) => r.id === sent.id && r.status === 'sent' && r.device === env.deviceId && r.sentAt === sent.sentAt,
(r) =>
final.ok
? { record: transition(r, { type: 'success' }, done), text: final.text }

7
src/core/schedule.ts
View File

@ -36,3 +36,10 @@ export function claimable(items: readonly Located[], now: Date, deviceId: string
r.device !== deviceId && (r.status === 'saved' || r.status === 'retry' || isStaleSent(r, now)),
);
}
// «Распознать все ожидающие здесь» также разбирает собственные retry
// (независимо от voice-next-attempt) и собственные зависшие sent — без
// ожидания, как и чужие claimable().
export function ownReclaimable(items: readonly Located[], now: Date, deviceId: string): Located[] {
return items.filter(({ record: r }) => r.device === deviceId && (r.status === 'retry' || isStaleSent(r, now)));
}

37
tests/queue-commands.test.ts
View File

@ -52,9 +52,13 @@ describe('QueueRunner.retry', () => {
expect(fields(ctx, path)).toMatchObject({ 'voice-status': 'inserted', 'voice-device': 'test-device' });
});
it.each(['saved', 'inserted'] as const)('%s: no change, no request', async (status) => {
it.each([
['saved', {}],
['inserted', {}],
['sent', { sentAt: T0.toISOString() }],
] as const)('%s: no change, no request', async (status, over) => {
const ctx = setup([]);
const path = seed(ctx, rec('a', { status }));
const path = seed(ctx, rec('a', { status, ...over }));
const before = ctx.notes.files.get(path);
expect(await ctx.runner.retry(path)).toBe(false);
expect(ctx.notes.files.get(path)).toBe(before);
@ -64,6 +68,16 @@ describe('QueueRunner.retry', () => {
it('missing note → false', async () => {
expect(await setup().runner.retry('voice/inbox/nope.md')).toBe(false);
});
it('after dispose(): no change, no request', async () => {
const ctx = setup([]);
const path = seed(ctx, rec('a', { status: 'failed', attempts: 5, error: 'e' }));
const before = ctx.notes.files.get(path);
ctx.runner.dispose();
expect(await ctx.runner.retry(path)).toBe(false);
expect(ctx.notes.files.get(path)).toBe(before);
expect(ctx.transcriber.calls).toHaveLength(0);
});
});
describe('QueueRunner.claimAll', () => {
@ -75,21 +89,30 @@ describe('QueueRunner.claimAll', () => {
expect(fields(ctx, path)).toMatchObject({ 'voice-status': 'inserted', 'voice-device': 'test-device', 'voice-attempts': '2' });
});
it('includes stale sent; leaves failed, inserted, fresh sent and own records', async () => {
const ctx = setup([{ ok: true, text: 'x' }, { ok: true, text: 'y' }]);
it('includes stale sent, own retry and own stale sent; leaves failed, inserted and fresh sent untouched', async () => {
const ctx = setup([
{ ok: true, text: 'w' },
{ ok: true, text: 'x' },
{ ok: true, text: 'y' },
{ ok: true, text: 'z' },
]);
const o = { device: 'iphone' };
const saved = seed(ctx, rec('a', o));
const stale = seed(ctx, rec('b', { ...o, status: 'sent', sentAt: '2026-09-27T11:04:00.000Z' }));
const fresh = seed(ctx, rec('c', { ...o, status: 'sent', sentAt: '2026-09-27T11:13:03.000Z' }));
const failed = seed(ctx, rec('d', { ...o, status: 'failed', attempts: 5, error: 'e' }));
const done = seed(ctx, rec('e', { ...o, status: 'inserted' }));
const own = seed(ctx, rec('f', { status: 'retry', attempts: 1, nextAttempt: '2026-09-27T11:35:03.000Z', error: 't' }));
const untouched = [fresh, failed, done, own].map((p) => ctx.notes.files.get(p));
const ownRetry = seed(ctx, rec('f', { status: 'retry', attempts: 1, nextAttempt: '2026-09-27T11:35:03.000Z', error: 't' }));
const ownStale = seed(ctx, rec('g', { status: 'sent', sentAt: '2026-09-27T11:04:00.000Z' }));
const ownFresh = seed(ctx, rec('h', { status: 'sent', sentAt: '2026-09-27T11:13:03.000Z' }));
const untouched = [fresh, failed, done, ownFresh].map((p) => ctx.notes.files.get(p));
expect(await ctx.runner.claimAll()).toBe(2);
expect(fields(ctx, saved)['voice-status']).toBe('inserted');
expect(fields(ctx, stale)['voice-status']).toBe('inserted');
expect([fresh, failed, done, own].map((p) => ctx.notes.files.get(p))).toEqual(untouched);
expect(fields(ctx, ownRetry)['voice-status']).toBe('inserted');
expect(fields(ctx, ownStale)['voice-status']).toBe('inserted');
expect([fresh, failed, done, ownFresh].map((p) => ctx.notes.files.get(p))).toEqual(untouched);
});
it('nothing to claim → 0', async () => {

51
tests/queue.test.ts
View File

@ -1,5 +1,5 @@
import { describe, expect, it } from 'vitest';
import { initialNote, readVoiceFields } from '../src/core/note';
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';
@ -259,4 +259,53 @@ describe('QueueRunner.tick', () => {
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');
});
});