fix(meeting-document): hold quota for in-flight generation claims
This commit is contained in:
Yun Chan
parent
f4724ddf53
commit
b35676c75c
5 changed files with 994 additions and 179 deletions
|
|
@ -0,0 +1,273 @@
|
|||
// Application core of generate-meeting-document.
|
||||
//
|
||||
// The flow (claim -> provider -> commit, with a failure marker on every exit
|
||||
// after a successful claim) depends only on the ports below, so it can be
|
||||
// exercised without Supabase, Anthropic or an HTTP runtime. index.ts is the
|
||||
// adapter that wires real clients into these ports and turns the result into a
|
||||
// Response.
|
||||
//
|
||||
// Quota: the claim RPC holds one unit of the user's allowance while the request
|
||||
// is 'processing' (migration 20260928000037). Every path that leaves a claimed
|
||||
// request unfinished must therefore call markFailed so the unit is released
|
||||
// immediately instead of waiting for the lease to expire.
|
||||
|
||||
import { buildMeetingDocumentSystemPrompt } from '../_shared/generative-ai-safety.ts'
|
||||
import {
|
||||
type GenerateMeetingDocumentRequest,
|
||||
parseProviderDocumentResult,
|
||||
type ProviderDocumentResult,
|
||||
} from '../_shared/meeting-document-contract.ts'
|
||||
|
||||
export type GenerationFailureCode =
|
||||
| 'provider_timeout'
|
||||
| 'provider_request_failed'
|
||||
| 'provider_invalid_response'
|
||||
| 'quota_exceeded'
|
||||
| 'commit_failed'
|
||||
|
||||
export interface DatabaseError {
|
||||
code?: string
|
||||
message?: string
|
||||
}
|
||||
|
||||
export interface RpcResult {
|
||||
data: unknown
|
||||
error: DatabaseError | null
|
||||
}
|
||||
|
||||
export interface GenerationHttpResult {
|
||||
status: number
|
||||
body: Record<string, unknown>
|
||||
}
|
||||
|
||||
export interface ClaimResult {
|
||||
claimed: boolean
|
||||
status: 'processing' | 'succeeded' | 'failed'
|
||||
documentId: string | null
|
||||
meetingTitle: string
|
||||
documentTitle: string
|
||||
templateType: string
|
||||
systemPrompt: string | null
|
||||
transcript: string | null
|
||||
model: string
|
||||
}
|
||||
|
||||
export interface ProviderRequest {
|
||||
model: string
|
||||
max_tokens: number
|
||||
system: string
|
||||
messages: Array<{ role: 'user'; content: string }>
|
||||
stream: false
|
||||
}
|
||||
|
||||
/** Outcome of one provider call, already classified by the adapter. */
|
||||
export type ProviderOutcome =
|
||||
| { kind: 'ok'; body: unknown }
|
||||
| { kind: 'timeout' }
|
||||
| { kind: 'network_error' }
|
||||
| { kind: 'http_error'; status: number }
|
||||
|
||||
/** Persistence port: the four RPC/table operations the flow needs. */
|
||||
export interface MeetingDocumentStore {
|
||||
claim(actorId: string, request: GenerateMeetingDocumentRequest): Promise<RpcResult>
|
||||
markFailed(actorId: string, idempotencyKey: string, code: GenerationFailureCode): Promise<DatabaseError | null>
|
||||
commit(
|
||||
actorId: string,
|
||||
idempotencyKey: string,
|
||||
result: ProviderDocumentResult,
|
||||
latencyMs: number,
|
||||
): Promise<RpcResult>
|
||||
findDocument(actorId: string, documentId: string): Promise<RpcResult>
|
||||
}
|
||||
|
||||
/** Provider port: sends one document request and classifies the transport outcome. */
|
||||
export interface MeetingDocumentProvider {
|
||||
generate(request: ProviderRequest): Promise<ProviderOutcome>
|
||||
}
|
||||
|
||||
export type GenerationLogger = (message: string, context?: Record<string, unknown>) => void
|
||||
|
||||
export interface GenerationDeps {
|
||||
store: MeetingDocumentStore
|
||||
provider: MeetingDocumentProvider
|
||||
now: () => number
|
||||
log: GenerationLogger
|
||||
}
|
||||
|
||||
export const PROVIDER_MAX_TOKENS = 4096
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === 'object' && value !== null && !Array.isArray(value)
|
||||
}
|
||||
|
||||
function result(status: number, body: Record<string, unknown>): GenerationHttpResult {
|
||||
return { status, body }
|
||||
}
|
||||
|
||||
export function parseClaim(value: unknown): ClaimResult | null {
|
||||
if (!isRecord(value)) return null
|
||||
if (
|
||||
typeof value.claimed !== 'boolean'
|
||||
|| !['processing', 'succeeded', 'failed'].includes(String(value.status))
|
||||
|| (value.documentId !== null && typeof value.documentId !== 'string')
|
||||
|| typeof value.meetingTitle !== 'string'
|
||||
|| typeof value.documentTitle !== 'string'
|
||||
|| typeof value.templateType !== 'string'
|
||||
|| (value.systemPrompt !== null && typeof value.systemPrompt !== 'string')
|
||||
|| (value.transcript !== null && typeof value.transcript !== 'string')
|
||||
|| typeof value.model !== 'string'
|
||||
) {
|
||||
return null
|
||||
}
|
||||
return value as unknown as ClaimResult
|
||||
}
|
||||
|
||||
export function isQuotaExceeded(error: DatabaseError | null): boolean {
|
||||
return error?.message?.includes('generation_quota_exceeded') === true
|
||||
}
|
||||
|
||||
export function mapDatabaseError(error: DatabaseError | null, log: GenerationLogger): GenerationHttpResult {
|
||||
const code = error?.code ?? ''
|
||||
if (code === '42501') return result(403, { error: 'forbidden' })
|
||||
if (code === 'P0002') return result(404, { error: 'not_found' })
|
||||
if (code === '22023') return result(400, { error: 'invalid_request' })
|
||||
if (isQuotaExceeded(error)) return result(429, { error: 'quota_exceeded' })
|
||||
log('meeting document database operation failed', { code: code || 'unknown' })
|
||||
return result(500, { error: 'internal_error' })
|
||||
}
|
||||
|
||||
export function buildProviderRequest(
|
||||
claim: ClaimResult & { systemPrompt: string; transcript: string },
|
||||
): ProviderRequest {
|
||||
return {
|
||||
model: claim.model,
|
||||
max_tokens: PROVIDER_MAX_TOKENS,
|
||||
system: buildMeetingDocumentSystemPrompt(claim.systemPrompt),
|
||||
messages: [{
|
||||
role: 'user',
|
||||
content: `회의 제목: ${claim.meetingTitle}\n\n전사록:\n${claim.transcript}`,
|
||||
}],
|
||||
stream: false,
|
||||
}
|
||||
}
|
||||
|
||||
async function replayClaim(
|
||||
actorId: string,
|
||||
claim: ClaimResult,
|
||||
deps: GenerationDeps,
|
||||
): Promise<GenerationHttpResult> {
|
||||
if (claim.status === 'succeeded' && claim.documentId !== null) {
|
||||
const { data: document, error } = await deps.store.findDocument(actorId, claim.documentId)
|
||||
if (error !== null || document === null || document === undefined) {
|
||||
deps.log('idempotent meeting document lookup failed', { code: error?.code ?? 'missing' })
|
||||
return result(500, { error: 'internal_error' })
|
||||
}
|
||||
return result(200, { document, idempotent: true })
|
||||
}
|
||||
return result(409, {
|
||||
error: claim.status === 'processing' ? 'generation_in_progress' : 'generation_failed',
|
||||
})
|
||||
}
|
||||
|
||||
async function runClaimedGeneration(
|
||||
actorId: string,
|
||||
request: GenerateMeetingDocumentRequest,
|
||||
claim: ClaimResult & { systemPrompt: string; transcript: string },
|
||||
deps: GenerationDeps,
|
||||
markFailed: (code: GenerationFailureCode) => Promise<void>,
|
||||
): Promise<GenerationHttpResult> {
|
||||
const startedAt = deps.now()
|
||||
const outcome = await deps.provider.generate(buildProviderRequest(claim))
|
||||
|
||||
if (outcome.kind === 'timeout') {
|
||||
await markFailed('provider_timeout')
|
||||
return result(504, { error: 'provider_timeout' })
|
||||
}
|
||||
if (outcome.kind === 'network_error') {
|
||||
await markFailed('provider_request_failed')
|
||||
return result(502, { error: 'provider_request_failed' })
|
||||
}
|
||||
if (outcome.kind === 'http_error') {
|
||||
deps.log('meeting document provider request failed', { status: outcome.status })
|
||||
await markFailed('provider_request_failed')
|
||||
return result(502, { error: 'provider_request_failed' })
|
||||
}
|
||||
|
||||
const providerResult = parseProviderDocumentResult(outcome.body)
|
||||
if (providerResult === null) {
|
||||
deps.log('meeting document provider returned an invalid response shape')
|
||||
await markFailed('provider_invalid_response')
|
||||
return result(502, { error: 'provider_invalid_response' })
|
||||
}
|
||||
|
||||
const { data: commitData, error: commitError } = await deps.store.commit(
|
||||
actorId,
|
||||
request.idempotencyKey,
|
||||
providerResult,
|
||||
Math.max(0, deps.now() - startedAt),
|
||||
)
|
||||
if (commitError !== null) {
|
||||
const quotaExceeded = isQuotaExceeded(commitError)
|
||||
await markFailed(quotaExceeded ? 'quota_exceeded' : 'commit_failed')
|
||||
if (quotaExceeded) return result(429, { error: 'quota_exceeded' })
|
||||
deps.log('meeting document atomic commit failed', { code: commitError.code ?? 'unknown' })
|
||||
return result(500, { error: 'commit_failed' })
|
||||
}
|
||||
if (!isRecord(commitData) || !isRecord(commitData.document)) {
|
||||
deps.log('meeting document commit returned an invalid shape')
|
||||
return result(500, { error: 'internal_error' })
|
||||
}
|
||||
|
||||
return result(200, {
|
||||
document: commitData.document,
|
||||
idempotent: commitData.idempotent === true,
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Claims the request, calls the provider and commits the document.
|
||||
* Throws only for unexpected adapter failures; a claimed request is marked
|
||||
* failed before such an error propagates, so its quota unit is released.
|
||||
*/
|
||||
export async function generateMeetingDocument(
|
||||
actorId: string,
|
||||
request: GenerateMeetingDocumentRequest,
|
||||
deps: GenerationDeps,
|
||||
): Promise<GenerationHttpResult> {
|
||||
const markFailed = async (code: GenerationFailureCode): Promise<void> => {
|
||||
try {
|
||||
const error = await deps.store.markFailed(actorId, request.idempotencyKey, code)
|
||||
if (error !== null) {
|
||||
deps.log('meeting document failure marker failed', { code: error.code ?? 'unknown' })
|
||||
}
|
||||
} catch {
|
||||
deps.log('meeting document failure marker failed', { code: 'exception' })
|
||||
}
|
||||
}
|
||||
|
||||
const { data: claimData, error: claimError } = await deps.store.claim(actorId, request)
|
||||
if (claimError !== null) return mapDatabaseError(claimError, deps.log)
|
||||
|
||||
const claim = parseClaim(claimData)
|
||||
if (claim === null) {
|
||||
deps.log('meeting document claim returned an invalid shape')
|
||||
return result(500, { error: 'internal_error' })
|
||||
}
|
||||
if (!claim.claimed) return await replayClaim(actorId, claim, deps)
|
||||
|
||||
if (claim.systemPrompt === null || claim.transcript === null) {
|
||||
await markFailed('commit_failed')
|
||||
return result(500, { error: 'internal_error' })
|
||||
}
|
||||
const readyClaim = { ...claim, systemPrompt: claim.systemPrompt, transcript: claim.transcript }
|
||||
|
||||
try {
|
||||
return await runClaimedGeneration(actorId, request, readyClaim, deps, markFailed)
|
||||
} catch (error) {
|
||||
// Release the in-flight quota unit held by this claim; the request is not
|
||||
// going to be committed. A no-op if commit already moved it out of
|
||||
// 'processing'.
|
||||
await markFailed('commit_failed')
|
||||
throw error
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue