feat: Wiederholregel und Voice-Claims (T7 L2 Task 10)

RetryPolicy mit retryDelayMsFor (bestehende Aufrufer unverändert),
Phasen transcribe/extract mit eigenen Wartezeiten (5/15 min bzw.
30 s/2 min, 3 Versuche) und Stale-Fenstern (30 bzw. 5 min). Claims per
FOR UPDATE SKIP LOCKED mit claimToken als Compare-and-Set, nur
Notizen mit gültigem Inhalt; Erfolg setzt den nächsten Status und nach
dem Abschreiben attempts = 0; Fehler meldet 'lost', wenn die Notiz
inzwischen bestätigt oder verworfen wurde; Aufschub mit Grund;
Stale-Freigabe transaktionsfähig mit den beendeten Notizen als
Rückgabe. Eine Quelle für die 14 Fehlercodes.

Refs #27

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
Nick SzaunigandClaude Opus 5.5 committed 2026-10-16 09:32:16 +02:00
1 parent af9df0afe4
commit 2a4dc868d9
8 files changed
+959 -3

No files matched your search

@@ -1,7 +1,9 @@
import {
DEFAULT_RETRY_POLICY,
MAX_ATTEMPTS,
RETRY_DELAYS_MS,
retryDelayMs,
retryDelayMsFor,
STALE_CLAIM_MS,
} from './retry-policy';
@@ -26,4 +28,29 @@ describe('retry policy', () => {
expect(() => retryDelayMs(attempts)).toThrow(RangeError);
},
);
it('keeps retryDelayMs identical to the default policy', () => {
expect(DEFAULT_RETRY_POLICY).toEqual({
delaysMs: RETRY_DELAYS_MS,
maxAttempts: MAX_ATTEMPTS,
staleClaimMs: STALE_CLAIM_MS,
});
for (const attempts of [1, 2, 3, 4, 5, 6, 9]) {
expect(retryDelayMsFor(DEFAULT_RETRY_POLICY, attempts)).toBe(
retryDelayMs(attempts),
);
}
});
it("retryDelayMsFor uses the policy's delays and repeats the last one", () => {
const policy = {
delaysMs: [1_000, 5_000],
maxAttempts: 4,
staleClaimMs: 60_000,
};
expect([1, 2, 3].map((n) => retryDelayMsFor(policy, n))).toEqual([
1_000, 5_000, 5_000,
]);
expect(() => retryDelayMsFor(policy, 0)).toThrow(RangeError);
});
});
+27 -3
View File
@@ -12,12 +12,36 @@ export const RETRY_DELAYS_MS = [
export const MAX_ATTEMPTS = 8;
export const STALE_CLAIM_MS = 300_000;
/** Delay after `attempts` failed attempts (1 = the first failure). */
export function retryDelayMs(attempts: number): number {
/**
* A retry rule as data, for queues with their own pace (voice notes: 3
* attempts, minutes instead of hours). `delaysMs[n - 1]` is the wait after
* the n-th failure, the last entry repeats; the `maxAttempts`-th failure
* ends in FAILED; a claim older than `staleClaimMs` belongs to a dead worker.
*/
export interface RetryPolicy {
readonly delaysMs: readonly number[];
readonly maxAttempts: number;
readonly staleClaimMs: number;
}
/** The rule of the outboxes (the constants above). */
export const DEFAULT_RETRY_POLICY: RetryPolicy = {
delaysMs: RETRY_DELAYS_MS,
maxAttempts: MAX_ATTEMPTS,
staleClaimMs: STALE_CLAIM_MS,
};
/** Delay after `attempts` failed attempts (1 = the first failure) under `policy`. */
export function retryDelayMsFor(policy: RetryPolicy, attempts: number): number {
if (!Number.isInteger(attempts) || attempts < 1) {
throw new RangeError(
`Retry delay needs a positive number of failed attempts, got ${attempts}`,
);
}
return RETRY_DELAYS_MS[Math.min(attempts, RETRY_DELAYS_MS.length) - 1];
return policy.delaysMs[Math.min(attempts, policy.delaysMs.length) - 1];
}
/** Delay after `attempts` failed attempts (1 = the first failure). */
export function retryDelayMs(attempts: number): number {
return retryDelayMsFor(DEFAULT_RETRY_POLICY, attempts);
}
@@ -0,0 +1,139 @@
import { Prisma } from '../../../generated/prisma';
import { PrismaService } from '../../prisma/prisma.service';
import {
claimVoiceCaptures,
deferVoiceClaim,
releaseStaleVoiceClaims,
settleVoiceFailure,
settleVoiceSuccess,
VoiceClaim,
} from './voice-claims';
function fakeDb() {
const queryRaw = jest.fn().mockResolvedValue([]);
const executeRaw = jest.fn().mockResolvedValue(1);
return {
db: {
$queryRaw: queryRaw,
$executeRaw: executeRaw,
} as unknown as PrismaService,
queryRaw,
executeRaw,
};
}
const claim: VoiceClaim = {
id: 'c1',
claimToken: 't1',
attempts: 0,
ownerUserId: 'u1',
colonyId: 'k1',
audioMime: 'audio/webm',
durationSec: 30,
};
function sqlOf(call: unknown): Prisma.Sql {
return (call as unknown[])[0] as Prisma.Sql;
}
describe('voice claims (SQL shape)', () => {
it('claims with SKIP LOCKED against the database clock', async () => {
const { db, queryRaw } = fakeDb();
await claimVoiceCaptures(db, 'transcribe', { limit: 5, maxTotalSec: 600 });
const text = sqlOf(queryRaw.mock.calls[0]).sql;
expect(text).toContain('FOR UPDATE SKIP LOCKED');
expect(text).toContain('now()');
expect(text).toContain('gen_random_uuid()');
});
it.each([0, -1, 1.5])('refuses the claim limit %p', async (limit) => {
const { db } = fakeDb();
await expect(claimVoiceCaptures(db, 'extract', { limit })).rejects.toThrow(
RangeError,
);
});
it('refuses an unknown phase', async () => {
const { db } = fakeDb();
await expect(
claimVoiceCaptures(db, 'drop' as never, { limit: 1 }),
).rejects.toThrow(/Unknown voice phase/);
});
it('settles by compare-and-set on status and claim token', async () => {
const { db, executeRaw } = fakeDb();
await expect(settleVoiceSuccess(db, 'transcribe', claim)).resolves.toBe(
true,
);
const { sql, values } = sqlOf(executeRaw.mock.calls[0]);
expect(sql).toContain('"claimToken"');
expect(values).toEqual(
expect.arrayContaining(['c1', 't1', 'TRANSCRIBING']),
);
});
it('reports a lost claim when no row changed', async () => {
const { db, executeRaw } = fakeDb();
executeRaw.mockResolvedValue(0);
await expect(settleVoiceSuccess(db, 'extract', claim)).resolves.toBe(false);
await expect(deferVoiceClaim(db, 'extract', claim, 1000)).resolves.toBe(
false,
);
});
it('decides retry or failed from the attempts before', async () => {
const { db } = fakeDb();
await expect(
settleVoiceFailure(
db,
'transcribe',
{ ...claim, attempts: 1 },
'TRANSCRIBE_FAILED',
),
).resolves.toBe('retry');
await expect(
settleVoiceFailure(
db,
'transcribe',
{ ...claim, attempts: 2 },
'TRANSCRIBE_FAILED',
),
).resolves.toBe('failed');
});
it('reports a lost failure settle', async () => {
const { db, executeRaw } = fakeDb();
executeRaw.mockResolvedValue(0);
await expect(
settleVoiceFailure(db, 'extract', claim, 'EXTRACT_RETRY'),
).resolves.toBe('lost');
});
it('turns the last EXTRACT_RETRY into EXTRACT_FAILED', async () => {
const { db, executeRaw } = fakeDb();
await settleVoiceFailure(
db,
'extract',
{ ...claim, attempts: 2 },
'EXTRACT_RETRY',
);
const { values } = sqlOf(executeRaw.mock.calls[0]);
expect(values).toContain('EXTRACT_FAILED');
expect(values).not.toContain('EXTRACT_RETRY');
});
it.each([-1, 1.5])('refuses the defer delay %p', async (delay) => {
const { db } = fakeDb();
await expect(deferVoiceClaim(db, 'extract', claim, delay)).rejects.toThrow(
RangeError,
);
});
it('releases stale claims with the phase stale window and attempt limit', async () => {
const { db, queryRaw } = fakeDb();
await releaseStaleVoiceClaims(db, 'extract');
const { values } = sqlOf(queryRaw.mock.calls[0]);
expect(values).toContain(300_000);
expect(values).toContain(3);
});
});
@@ -0,0 +1,304 @@
import { Prisma } from '../../../generated/prisma';
import { retryDelayMsFor } from '../../common/retry-policy';
import { DbClient } from '../../prisma/db-client';
import { PrismaService } from '../../prisma/prisma.service';
import { VoiceErrorCode } from '../voice-error-codes';
import { VOICE_PHASES, VoicePhase } from './voice-phases';
/**
* Claim, settle and stale release for voice notes - the same SQL form as
* outbox-claims.ts (SKIP LOCKED, database clock, closed table map), but with
* two phases that have their own states and retry rule, and a claimToken
* instead of claimedAt as the compare-and-set key: the claim travels across
* processes (fetch -> whisper -> ingest), so the key must be a value both
* halves hold. Settling is a compare-and-set on status + claimToken; a note
* that was confirmed, discarded or released in the meantime is not touched.
* Part 4 adds `VoiceUtterance` to VOICE_CLAIM_TABLES.
*/
export type VoiceClaimTable = 'VoiceCapture';
interface VoiceClaimTableSql {
table: Prisma.Sql;
statusType: Prisma.Sql;
}
/**
* Table and enum names cannot be bound as parameters. They come only from
* this closed map - never from caller input - which is what makes
* Prisma.raw safe here.
*/
export const VOICE_CLAIM_TABLES: Readonly<
Record<VoiceClaimTable, VoiceClaimTableSql>
> = {
VoiceCapture: {
table: Prisma.raw('"VoiceCapture"'),
statusType: Prisma.raw('"VoiceCaptureStatus"'),
},
};
const TABLE = VOICE_CLAIM_TABLES.VoiceCapture;
/** Longest wait deferVoiceClaim allows: a day. */
const MAX_DEFER_MS = 24 * 60 * 60 * 1000;
/** What a claimed note gives its worker: no content, only what the job needs. */
export interface VoiceClaim {
id: string;
claimToken: string;
/** Attempts before this claim (0 on the first). */
attempts: number;
ownerUserId: string;
colonyId: string;
audioMime: string | null;
durationSec: number | null;
}
export interface VoiceClaimOptions {
limit: number;
/** Total reported seconds of audio per call; the first note is always claimed. */
maxTotalSec?: number;
/**
* Extra condition, ANDed to the claim (e.g. an owner filter in tests).
* Trusted SQL only: never build it with Prisma.raw from input.
*/
filter?: Prisma.Sql;
}
/** Codes a job may report for a failed attempt. */
export type VoiceFailureCode =
'TRANSCRIBE_FAILED' | 'EXTRACT_RETRY' | 'EXTRACT_INVALID';
/** Codes shown while a note waits without having failed. */
export type VoiceDeferCode = 'EXTRACT_DISABLED' | 'DAILY_LIMIT';
/** A note a stale release touched; `code` is set when it ended FAILED. */
export interface ReleasedVoiceClaim {
id: string;
status: string;
code: VoiceErrorCode | null;
}
function phaseOf(phase: VoicePhase) {
if (!Object.hasOwn(VOICE_PHASES, phase)) {
throw new Error(`Unknown voice phase "${String(phase)}"`);
}
return VOICE_PHASES[phase];
}
function toClaim(row: VoiceClaim): VoiceClaim {
return {
id: row.id,
claimToken: row.claimToken,
attempts: row.attempts,
ownerUserId: row.ownerUserId,
colonyId: row.colonyId,
audioMime: row.audioMime,
durationSec: row.durationSec,
};
}
/**
* Claims up to `limit` due notes of the phase whose content has not
* expired, oldest first, each with its own fresh claimToken, and - with
* maxTotalSec - only as many as fit the audio budget (a running sum over
* the locked rows; the first note is always taken so a long note cannot
* starve). "availableAt" is TIMESTAMP(3): compare against now() rounded the
* same way, see claimBatch. Uses the root client: a claim must commit on
* its own.
*/
export async function claimVoiceCaptures(
db: PrismaService,
phase: VoicePhase,
options: VoiceClaimOptions,
): Promise<VoiceClaim[]> {
const { from, working } = phaseOf(phase);
if (!Number.isInteger(options.limit) || options.limit < 1) {
throw new RangeError(
`Claim limit must be a positive integer, got ${options.limit}`,
);
}
const budget = options.maxTotalSec;
if (budget !== undefined && (!Number.isInteger(budget) || budget < 1)) {
throw new RangeError(
`Claim audio budget must be a positive integer, got ${budget}`,
);
}
const filter = options.filter
? Prisma.sql`AND ${options.filter}`
: Prisma.empty;
const fits =
budget === undefined
? Prisma.empty
: Prisma.sql`WHERE "position" = 1 OR "runningSec" <= ${budget}`;
const rows = await db.$queryRaw<VoiceClaim[]>(Prisma.sql`
WITH locked AS (
SELECT "id", "createdAt", COALESCE("durationSec", 0) AS "sec"
FROM ${TABLE.table}
WHERE "status" = CAST(${from} AS ${TABLE.statusType})
AND "availableAt" <= CAST(now() AS timestamp(3))
AND "contentExpiresAt" > now() ${filter}
ORDER BY "createdAt", "id"
LIMIT ${options.limit}
FOR UPDATE SKIP LOCKED
), ranked AS (
SELECT "id",
row_number() OVER (ORDER BY "createdAt", "id") AS "position",
sum("sec") OVER (ORDER BY "createdAt", "id") AS "runningSec"
FROM locked
), claimed AS (
UPDATE ${TABLE.table}
SET "status" = CAST(${working} AS ${TABLE.statusType}),
"claimedAt" = now(),
"updatedAt" = now(),
"claimToken" = gen_random_uuid()::text
WHERE "id" IN (SELECT "id" FROM ranked ${fits})
RETURNING "id", "claimToken", "attempts", "ownerUserId", "colonyId",
"audioMime", "durationSec", "createdAt"
)
SELECT "id", "claimToken", "attempts", "ownerUserId", "colonyId",
"audioMime", "durationSec"
FROM claimed ORDER BY "createdAt", "id"`);
return rows.map(toClaim);
}
/**
* Settles a claimed note as done: sets the phase's next status
* (TRANSCRIBED or READY), releases the claim and - after transcribing -
* resets attempts to 0 (Ruling 6), so no caller has to remember it. `set`
* holds only the result columns (e.g. `"transcriptEnc" = ...`); it must not
* name status, attempts, claimedAt, claimToken or updatedAt. Pass the
* transaction client to settle together with the result's other writes.
* Returns false if the claim was lost (released as stale, claimed again,
* confirmed or discarded): the current state wins and nothing changed.
*/
export async function settleVoiceSuccess(
tx: DbClient,
phase: VoicePhase,
claim: Pick<VoiceClaim, 'id' | 'claimToken'>,
set?: Prisma.Sql,
): Promise<boolean> {
const { working, done } = phaseOf(phase);
const result = set ? Prisma.sql`${set},` : Prisma.empty;
const attempts =
phase === 'transcribe' ? Prisma.sql`"attempts" = 0,` : Prisma.empty;
const count = await tx.$executeRaw(Prisma.sql`
UPDATE ${TABLE.table}
SET ${result} ${attempts}
"status" = CAST(${done} AS ${TABLE.statusType}),
"updatedAt" = now(),
"claimedAt" = NULL,
"claimToken" = NULL
WHERE "id" = ${claim.id}
AND "status" = CAST(${working} AS ${TABLE.statusType})
AND "claimToken" = ${claim.claimToken}`);
return count === 1;
}
/**
* Settles a claimed note after a failed attempt: back to the phase's
* waiting state after the policy delay (database clock), or FAILED once the
* policy's maxAttempts is reached. A last EXTRACT_RETRY ends as
* EXTRACT_FAILED. Returns 'lost' (nothing changed) if the claim is gone: the
* note was confirmed or discarded meanwhile, or a stale release already
* counted this attempt and decided its fate.
*/
export async function settleVoiceFailure(
db: DbClient,
phase: VoicePhase,
claim: VoiceClaim,
code: VoiceFailureCode,
): Promise<'retry' | 'failed' | 'lost'> {
const { from, working, failureCode, policy } = phaseOf(phase);
const attempts = claim.attempts + 1;
const failed = attempts >= policy.maxAttempts;
const storedCode = failed && code === 'EXTRACT_RETRY' ? failureCode : code;
const next = failed
? Prisma.sql`"status" = CAST('FAILED' AS ${TABLE.statusType})`
: Prisma.sql`"status" = CAST(${from} AS ${TABLE.statusType}),
"availableAt" = now() + (${retryDelayMsFor(policy, attempts)} * interval '1 millisecond')`;
const count = await db.$executeRaw(Prisma.sql`
UPDATE ${TABLE.table}
SET ${next},
"attempts" = "attempts" + 1,
"updatedAt" = now(),
"claimedAt" = NULL,
"claimToken" = NULL,
"lastErrorCode" = ${storedCode}
WHERE "id" = ${claim.id}
AND "status" = CAST(${working} AS ${TABLE.statusType})
AND "claimToken" = ${claim.claimToken}`);
if (count !== 1) {
return 'lost';
}
return failed ? 'failed' : 'retry';
}
/**
* Puts a claimed note back to the phase's waiting state, due after
* `delayMs` by the database clock, without counting an attempt: it did not
* fail, it has to wait (daily limit, key missing). `code` is stored in
* lastErrorCode so the user sees why meanwhile. Capped at 24 hours. Returns
* false if the claim was lost.
*/
export async function deferVoiceClaim(
db: DbClient,
phase: VoicePhase,
claim: Pick<VoiceClaim, 'id' | 'claimToken'>,
delayMs: number,
code?: VoiceDeferCode,
): Promise<boolean> {
const { from, working } = phaseOf(phase);
if (!Number.isInteger(delayMs) || delayMs < 0) {
throw new RangeError(
`Defer delay must be a non-negative integer, got ${delayMs}`,
);
}
const lastError =
code === undefined ? Prisma.empty : Prisma.sql`"lastErrorCode" = ${code},`;
const count = await db.$executeRaw(Prisma.sql`
UPDATE ${TABLE.table}
SET ${lastError}
"status" = CAST(${from} AS ${TABLE.statusType}),
"updatedAt" = now(),
"claimedAt" = NULL,
"claimToken" = NULL,
"availableAt" = now() + (${Math.min(delayMs, MAX_DEFER_MS)} * interval '1 millisecond')
WHERE "id" = ${claim.id}
AND "status" = CAST(${working} AS ${TABLE.statusType})
AND "claimToken" = ${claim.claimToken}`);
return count === 1;
}
/**
* Puts notes of the phase whose claim is older than the phase's stale
* window (their worker died) back to the waiting state right away with
* attempts + 1 - or to FAILED (phase failure code) once that reaches
* maxAttempts, so a note that kills its worker cannot loop. Takes any
* client so the release and the VoiceCaptureFailed events of the FAILED
* ones (status FAILED, `code` set) can share one transaction.
*/
export async function releaseStaleVoiceClaims(
db: DbClient,
phase: VoicePhase,
options: { filter?: Prisma.Sql } = {},
): Promise<ReleasedVoiceClaim[]> {
const { from, working, failureCode, policy } = phaseOf(phase);
const filter = options.filter
? Prisma.sql`AND ${options.filter}`
: Prisma.empty;
return db.$queryRaw<ReleasedVoiceClaim[]>(Prisma.sql`
UPDATE ${TABLE.table}
SET "status" = CAST(
CASE WHEN "attempts" + 1 >= ${policy.maxAttempts} THEN 'FAILED' ELSE ${from} END
AS ${TABLE.statusType}),
"attempts" = "attempts" + 1,
"updatedAt" = now(),
"claimedAt" = NULL,
"claimToken" = NULL,
"availableAt" = now(),
"lastErrorCode" = CASE WHEN "attempts" + 1 >= ${policy.maxAttempts} THEN ${failureCode} ELSE "lastErrorCode" END
WHERE "status" = CAST(${working} AS ${TABLE.statusType})
AND "claimedAt" < now() - (${policy.staleClaimMs} * interval '1 millisecond') ${filter}
RETURNING "id", "status"::text AS "status",
CASE WHEN "status" = 'FAILED' THEN "lastErrorCode" END AS "code"`);
}
@@ -0,0 +1,27 @@
import { retryDelayMsFor } from '../../common/retry-policy';
import { VOICE_PHASES } from './voice-phases';
describe('voice phases', () => {
it('transcribes QUEUED to TRANSCRIBING with 3 attempts, 5 min / 15 min and a 30 min stale claim', () => {
const phase = VOICE_PHASES.transcribe;
expect([phase.from, phase.working]).toEqual(['QUEUED', 'TRANSCRIBING']);
expect(phase.policy).toEqual({
delaysMs: [300_000, 900_000],
maxAttempts: 3,
staleClaimMs: 1_800_000,
});
expect([1, 2].map((n) => retryDelayMsFor(phase.policy, n))).toEqual([
300_000, 900_000,
]);
});
it('extracts TRANSCRIBED to EXTRACTING with 3 attempts, 30 s / 2 min and a 5 min stale claim', () => {
const phase = VOICE_PHASES.extract;
expect([phase.from, phase.working]).toEqual(['TRANSCRIBED', 'EXTRACTING']);
expect(phase.policy).toEqual({
delaysMs: [30_000, 120_000],
maxAttempts: 3,
staleClaimMs: 300_000,
});
});
});
@@ -0,0 +1,44 @@
import { RetryPolicy } from '../../common/retry-policy';
import { VoiceErrorCode } from '../voice-error-codes';
/**
* The two job phases of a voice note. Each has its own states, its own
* retry rule and its own code when the last attempt fails. Both phases share
* one `attempts` counter; settling the transcribe phase resets it to 0
* (Ruling 6), so the extract phase starts with its own 3 attempts.
*/
export const VOICE_PHASES = {
transcribe: {
from: 'QUEUED',
working: 'TRANSCRIBING',
done: 'TRANSCRIBED',
failureCode: 'TRANSCRIBE_FAILED',
policy: {
delaysMs: [300_000, 900_000],
maxAttempts: 3,
staleClaimMs: 1_800_000,
},
},
extract: {
from: 'TRANSCRIBED',
working: 'EXTRACTING',
done: 'READY',
failureCode: 'EXTRACT_FAILED',
policy: {
delaysMs: [30_000, 120_000],
maxAttempts: 3,
staleClaimMs: 300_000,
},
},
} as const satisfies Record<
string,
{
from: string;
working: string;
done: string;
failureCode: VoiceErrorCode;
policy: RetryPolicy;
}
>;
export type VoicePhase = keyof typeof VOICE_PHASES;
@@ -0,0 +1,22 @@
/**
* The fixed error codes of a voice note. They land in lastErrorCode, in
* events and in API answers - never content, only one of these.
*/
export const VOICE_ERROR_CODES = [
'UNSUPPORTED_AUDIO',
'TOO_LARGE',
'TOO_MANY_OPEN',
'NO_SPEECH',
'TOO_LONG',
'TRANSCRIBE_FAILED',
'AUDIO_EXPIRED',
'EXTRACT_DISABLED',
'EXTRACT_RETRY',
'EXTRACT_FAILED',
'DAILY_LIMIT',
'EXTRACT_INVALID',
'REFUSED',
'VOICE_CAPTURE_GONE',
] as const;
export type VoiceErrorCode = (typeof VOICE_ERROR_CODES)[number];