hard-cut batch A: worker-only app compute backend
This commit is contained in:
parent
821d0fdc6b
commit
f5eb593554
7 changed files with 18 additions and 137 deletions
|
|
@ -1,24 +1,10 @@
|
||||||
import type { ComputeBackend, ComputeMode } from '@/lib/server/compute/types';
|
import type { ComputeBackend } from '@/lib/server/compute/types';
|
||||||
import { isComputeModeAvailable, readComputeMode } from '@/lib/server/compute/mode';
|
import { isWorkerClientConfigAvailable, WorkerComputeBackend } from '@/lib/server/compute/worker';
|
||||||
import { WorkerComputeBackend } from '@/lib/server/compute/worker';
|
|
||||||
|
|
||||||
declare const __OPENREADER_COMPUTE_MODE__: 'local' | 'worker' | 'none';
|
|
||||||
|
|
||||||
let backendPromise: Promise<ComputeBackend> | null = null;
|
let backendPromise: Promise<ComputeBackend> | null = null;
|
||||||
|
|
||||||
async function createBackend(): Promise<ComputeBackend> {
|
async function createBackend(): Promise<ComputeBackend> {
|
||||||
const bundledMode =
|
return new WorkerComputeBackend();
|
||||||
typeof __OPENREADER_COMPUTE_MODE__ === 'undefined' ? 'none' : __OPENREADER_COMPUTE_MODE__;
|
|
||||||
|
|
||||||
if (bundledMode === 'worker') return new WorkerComputeBackend();
|
|
||||||
if (bundledMode === 'local') {
|
|
||||||
const { LocalComputeBackend } = await import('@/lib/server/compute/local');
|
|
||||||
return new LocalComputeBackend();
|
|
||||||
}
|
|
||||||
const mode: ComputeMode = readComputeMode();
|
|
||||||
if (mode === 'worker') return new WorkerComputeBackend();
|
|
||||||
const { LocalComputeBackend } = await import('@/lib/server/compute/local');
|
|
||||||
return new LocalComputeBackend();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function getCompute(): Promise<ComputeBackend> {
|
export async function getCompute(): Promise<ComputeBackend> {
|
||||||
|
|
@ -32,5 +18,5 @@ export async function getCompute(): Promise<ComputeBackend> {
|
||||||
}
|
}
|
||||||
|
|
||||||
export function isComputeAvailable(): boolean {
|
export function isComputeAvailable(): boolean {
|
||||||
return isComputeModeAvailable();
|
return isWorkerClientConfigAvailable();
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -1,20 +0,0 @@
|
||||||
import type { ComputeBackend } from '@/lib/server/compute/types';
|
|
||||||
import { WorkerComputeBackend } from '@/lib/server/compute/worker';
|
|
||||||
|
|
||||||
let backendPromise: Promise<ComputeBackend> | null = null;
|
|
||||||
|
|
||||||
export async function getCompute(): Promise<ComputeBackend> {
|
|
||||||
if (!backendPromise) {
|
|
||||||
backendPromise = Promise.resolve()
|
|
||||||
.then(() => new WorkerComputeBackend())
|
|
||||||
.catch((error) => {
|
|
||||||
backendPromise = null;
|
|
||||||
throw error;
|
|
||||||
});
|
|
||||||
}
|
|
||||||
return backendPromise;
|
|
||||||
}
|
|
||||||
|
|
||||||
export function isComputeAvailable(): boolean {
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
|
|
@ -1,77 +0,0 @@
|
||||||
import type { ComputeBackend, PdfLayoutInput, WhisperAlignInput, WhisperAlignResult } from '@/lib/server/compute/types';
|
|
||||||
import { LOCAL_COMPUTE_LIMITER } from '@/lib/server/compute/concurrency-limiter';
|
|
||||||
import { getDocumentBlob } from '@/lib/server/documents/blobstore';
|
|
||||||
import { getTtsSegmentAudioObject } from '@/lib/server/tts/segments-blobstore';
|
|
||||||
import {
|
|
||||||
getComputeTimeoutConfig,
|
|
||||||
withIdleTimeoutAndHardCap,
|
|
||||||
withTimeout,
|
|
||||||
} from '@openreader/compute-core';
|
|
||||||
import type { PdfLayoutProgress } from '@openreader/compute-core/api-contracts';
|
|
||||||
import {
|
|
||||||
runPdfLayoutFromPdfBuffer,
|
|
||||||
runWhisperAlignmentFromAudioBuffer,
|
|
||||||
} from '@openreader/compute-core/local-runtime';
|
|
||||||
|
|
||||||
export class LocalComputeBackend implements ComputeBackend {
|
|
||||||
readonly mode = 'local' as const;
|
|
||||||
private readonly timeoutConfig = getComputeTimeoutConfig();
|
|
||||||
|
|
||||||
async alignWords(input: WhisperAlignInput): Promise<WhisperAlignResult> {
|
|
||||||
return LOCAL_COMPUTE_LIMITER.run(async () => {
|
|
||||||
let audioBuffer = input.audioBuffer ?? null;
|
|
||||||
if (!audioBuffer && input.audioObjectKey) {
|
|
||||||
const bytes = new Uint8Array(await getTtsSegmentAudioObject(input.audioObjectKey));
|
|
||||||
audioBuffer = bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength);
|
|
||||||
}
|
|
||||||
if (!audioBuffer) {
|
|
||||||
throw new Error('Local compute alignment requires audioBuffer or audioObjectKey');
|
|
||||||
}
|
|
||||||
return withTimeout(
|
|
||||||
runWhisperAlignmentFromAudioBuffer({
|
|
||||||
audioBuffer,
|
|
||||||
text: input.text,
|
|
||||||
cacheKey: input.cacheKey,
|
|
||||||
lang: input.lang,
|
|
||||||
}),
|
|
||||||
this.timeoutConfig.whisperTimeoutMs,
|
|
||||||
'local whisper alignment job',
|
|
||||||
);
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
async parsePdfLayout(input: PdfLayoutInput) {
|
|
||||||
return LOCAL_COMPUTE_LIMITER.run(async () => {
|
|
||||||
let pdfBytes = input.pdfBytes ?? null;
|
|
||||||
if (!pdfBytes && input.documentId && typeof input.namespace !== 'undefined') {
|
|
||||||
const bytes = new Uint8Array(await getDocumentBlob(input.documentId, input.namespace));
|
|
||||||
pdfBytes = bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength);
|
|
||||||
}
|
|
||||||
if (!pdfBytes) {
|
|
||||||
throw new Error('Local compute PDF layout requires pdfBytes or (documentId + namespace)');
|
|
||||||
}
|
|
||||||
|
|
||||||
return {
|
|
||||||
parsed: (await withIdleTimeoutAndHardCap({
|
|
||||||
idleTimeoutMs: Math.max(this.timeoutConfig.pdfTimeoutMs, 1_000),
|
|
||||||
hardCapMs: this.timeoutConfig.pdfHardCapMs,
|
|
||||||
label: 'local pdf layout job',
|
|
||||||
run: async (touchProgress) => runPdfLayoutFromPdfBuffer({
|
|
||||||
documentId: input.documentId,
|
|
||||||
pdfBytes,
|
|
||||||
onPageParsed: (page) => {
|
|
||||||
touchProgress();
|
|
||||||
const progress: PdfLayoutProgress = {
|
|
||||||
totalPages: page.totalPages,
|
|
||||||
pagesParsed: page.pageNumber,
|
|
||||||
currentPage: page.pageNumber,
|
|
||||||
phase: 'infer',
|
|
||||||
};
|
|
||||||
return input.onProgress?.(progress);
|
|
||||||
},
|
|
||||||
}),
|
|
||||||
})).parsed,
|
|
||||||
};
|
|
||||||
});
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,17 +0,0 @@
|
||||||
import type { ComputeMode } from '@/lib/server/compute/types';
|
|
||||||
|
|
||||||
export function readComputeMode(): ComputeMode {
|
|
||||||
const envValue = process.env.COMPUTE_MODE;
|
|
||||||
const raw = (envValue || '').trim().toLowerCase();
|
|
||||||
if (!raw) return 'local';
|
|
||||||
if (raw === 'local' || raw === 'worker') return raw;
|
|
||||||
throw new Error(
|
|
||||||
`Invalid COMPUTE_MODE="${envValue}". Expected "local" or "worker".`,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
export function isComputeModeAvailable(): boolean {
|
|
||||||
// Throws on invalid values so startup/runtime config fails fast.
|
|
||||||
readComputeMode();
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
|
|
@ -6,7 +6,7 @@ import type {
|
||||||
import type { PdfLayoutProgress, WhisperAlignJobBase } from '@openreader/compute-core/api-contracts';
|
import type { PdfLayoutProgress, WhisperAlignJobBase } from '@openreader/compute-core/api-contracts';
|
||||||
import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts';
|
import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts';
|
||||||
|
|
||||||
export type ComputeMode = 'local' | 'worker';
|
export type ComputeMode = 'worker';
|
||||||
|
|
||||||
export interface WhisperAlignInput extends WhisperAlignJobBase {
|
export interface WhisperAlignInput extends WhisperAlignJobBase {
|
||||||
audioBuffer?: TTSAudioBuffer;
|
audioBuffer?: TTSAudioBuffer;
|
||||||
|
|
@ -32,7 +32,7 @@ export type PdfLayoutResult =
|
||||||
| { parsed?: never; parsedObjectKey: string };
|
| { parsed?: never; parsedObjectKey: string };
|
||||||
|
|
||||||
export interface ComputeBackend {
|
export interface ComputeBackend {
|
||||||
mode: ComputeMode;
|
mode: 'worker';
|
||||||
alignWords(input: WhisperAlignInput): Promise<WhisperAlignResult>;
|
alignWords(input: WhisperAlignInput): Promise<WhisperAlignResult>;
|
||||||
parsePdfLayout(input: PdfLayoutInput): Promise<PdfLayoutResult>;
|
parsePdfLayout(input: PdfLayoutInput): Promise<PdfLayoutResult>;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -93,7 +93,7 @@ function opSummary(value: unknown): Record<string, unknown> {
|
||||||
|
|
||||||
function readRequiredEnv(name: string): string {
|
function readRequiredEnv(name: string): string {
|
||||||
const value = process.env[name]?.trim();
|
const value = process.env[name]?.trim();
|
||||||
if (!value) throw new Error(`${name} is required when COMPUTE_MODE=worker`);
|
if (!value) throw new Error(`${name} is required for compute worker client`);
|
||||||
return value;
|
return value;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -127,6 +127,15 @@ export function getWorkerClientConfigFromEnv(): { baseUrl: string; token: string
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export function isWorkerClientConfigAvailable(): boolean {
|
||||||
|
try {
|
||||||
|
getWorkerClientConfigFromEnv();
|
||||||
|
return true;
|
||||||
|
} catch {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
function parseRetryAfterMs(value: string | null): number | null {
|
function parseRetryAfterMs(value: string | null): number | null {
|
||||||
if (!value) return null;
|
if (!value) return null;
|
||||||
const asNum = Number(value);
|
const asNum = Number(value);
|
||||||
|
|
|
||||||
|
|
@ -6,7 +6,7 @@ import {
|
||||||
type RuntimeConfigKey,
|
type RuntimeConfigKey,
|
||||||
type RuntimeConfigSource,
|
type RuntimeConfigSource,
|
||||||
} from '@/lib/server/admin/settings';
|
} from '@/lib/server/admin/settings';
|
||||||
import { isComputeModeAvailable } from '@/lib/server/compute/mode';
|
import { isComputeAvailable } from '@/lib/server/compute';
|
||||||
|
|
||||||
export type ResolvedRuntimeConfig = RuntimeConfig & {
|
export type ResolvedRuntimeConfig = RuntimeConfig & {
|
||||||
computeAvailable: boolean;
|
computeAvailable: boolean;
|
||||||
|
|
@ -29,7 +29,7 @@ export async function getResolvedRuntimeConfig(): Promise<ResolvedRuntimeConfig>
|
||||||
const values = await getRuntimeConfig();
|
const values = await getRuntimeConfig();
|
||||||
return {
|
return {
|
||||||
...values,
|
...values,
|
||||||
computeAvailable: isComputeModeAvailable(),
|
computeAvailable: isComputeAvailable(),
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue