From 972976b16a58933dd15e71277f7c7d9edce024ac Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 27 Sep 2026 17:12:31 +0400 Subject: [PATCH] fix(core): claimAll reclaims own work; isolate per-record queue errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- src/core/queue.ts | 60 ++++++++++++++++++++++++++++++------ src/core/schedule.ts | 7 +++++ tests/queue-commands.test.ts | 37 +++++++++++++++++----- tests/queue.test.ts | 51 +++++++++++++++++++++++++++++- 4 files changed, 138 insertions(+), 17 deletions(-) diff --git a/src/core/queue.ts b/src/core/queue.ts index d69d1ec..e369b70 100644 --- a/src/core/queue.ts +++ b/src/core/queue.ts @@ -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 { if (this.controller.signal.aborted) return; if (this.running) { @@ -50,6 +53,7 @@ export class QueueRunner { // «Повторить распознавание»: failed/retry → saved, счётчик с нуля, запись // переходит текущему устройству (иначе чужую запись здесь не отправить). async retry(path: string): Promise { + 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 { + 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 { 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 { @@ -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 } diff --git a/src/core/schedule.ts b/src/core/schedule.ts index 4e51ab2..63f5ca4 100644 --- a/src/core/schedule.ts +++ b/src/core/schedule.ts @@ -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))); +} diff --git a/tests/queue-commands.test.ts b/tests/queue-commands.test.ts index 8d6645a..53440b1 100644 --- a/tests/queue-commands.test.ts +++ b/tests/queue-commands.test.ts @@ -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 () => { diff --git a/tests/queue.test.ts b/tests/queue.test.ts index 2dc8af5..ecee73d 100644 --- a/tests/queue.test.ts +++ b/tests/queue.test.ts @@ -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'); + }); });