diff --git a/src/app/api/tts/route.ts b/src/app/api/tts/route.ts index 52caa14..d22c817 100644 --- a/src/app/api/tts/route.ts +++ b/src/app/api/tts/route.ts @@ -23,6 +23,87 @@ const ttsAudioCache = new LRUCache({ ttl: TTS_CACHE_TTL_MS, }); +// Concurrency controls and in-flight de-duplication +const TTS_MAX_CONCURRENCY = Number(process.env.TTS_MAX_CONCURRENCY || 4); + +class Semaphore { + private permits: number; + private queue: Array<() => void> = []; + constructor(max: number) { + this.permits = Math.max(1, max); + } + async acquire(): Promise<() => void> { + if (this.permits > 0) { + this.permits -= 1; + return this.release.bind(this); + } + return new Promise<() => void>((resolve) => { + this.queue.push(() => { + this.permits -= 1; + resolve(this.release.bind(this)); + }); + }); + } + private release() { + this.permits += 1; + const next = this.queue.shift(); + if (next) next(); + } +} + +const ttsSemaphore = new Semaphore(TTS_MAX_CONCURRENCY); + +type InflightEntry = { + promise: Promise; + controller: AbortController; + consumers: number; +}; + +const inflightRequests = new Map(); + +function sleep(ms: number) { + return new Promise((res) => setTimeout(res, ms)); +} + +async function fetchTTSBufferWithRetry( + openai: OpenAI, + createParams: ExtendedSpeechParams, + signal: AbortSignal +): Promise { + let attempt = 0; + const maxRetries = Number(process.env.TTS_MAX_RETRIES ?? 2); + let delay = Number(process.env.TTS_RETRY_INITIAL_MS ?? 250); + const maxDelay = Number(process.env.TTS_RETRY_MAX_MS ?? 2000); + const backoff = Number(process.env.TTS_RETRY_BACKOFF ?? 2); + + // Retry on 429 and 5xx only; never retry aborts + for (;;) { + try { + const response = await openai.audio.speech.create(createParams as SpeechCreateParams, { signal }); + return await response.arrayBuffer(); + } catch (err: unknown) { + if (signal?.aborted || (err instanceof Error && err.name === 'AbortError')) { + throw err; + } + const status = (() => { + if (typeof err === 'object' && err !== null) { + const rec = err as Record; + if (typeof rec.status === 'number') return rec.status as number; + if (typeof rec.statusCode === 'number') return rec.statusCode as number; + } + return 0; + })(); + const retryable = status === 429 || status >= 500; + if (!retryable || attempt >= maxRetries) { + throw err; + } + await sleep(Math.min(delay, maxDelay)); + delay = Math.min(maxDelay, delay * backoff); + attempt += 1; + } + } +} + function makeCacheKey(input: { provider: string; model: string | null | undefined; @@ -102,30 +183,108 @@ export async function POST(req: NextRequest) { instructions: createParams.instructions, }); + const etag = `W/"${cacheKey}"`; + const ifNoneMatch = req.headers.get('if-none-match'); + const cachedBuffer = ttsAudioCache.get(cacheKey); if (cachedBuffer) { + if (ifNoneMatch && (ifNoneMatch.includes(cacheKey) || ifNoneMatch.includes(etag))) { + return new NextResponse(null, { + status: 304, + headers: { + 'ETag': etag, + 'Cache-Control': 'private, max-age=1800', + 'Vary': 'x-tts-provider, x-openai-key, x-openai-base-url' + } + }); + } console.log('TTS cache HIT for key:', cacheKey.slice(0, 8)); return new NextResponse(cachedBuffer, { headers: { 'Content-Type': contentType, 'X-Cache': 'HIT', + 'ETag': etag, + 'Content-Length': String(cachedBuffer.byteLength), + 'Cache-Control': 'private, max-age=1800', + 'Vary': 'x-tts-provider, x-openai-key, x-openai-base-url' } }); } - const response = await openai.audio.speech.create(createParams as SpeechCreateParams, { signal: req.signal }); + // De-duplicate identical in-flight requests and bound upstream concurrency + const existing = inflightRequests.get(cacheKey); + if (existing) { + console.log('TTS in-flight JOIN for key:', cacheKey.slice(0, 8)); + existing.consumers += 1; - // Read the audio data as an ArrayBuffer and return it with appropriate headers - // This will also be aborted if the client cancels - const buffer = await response.arrayBuffer(); + const onAbort = (_evt: Event) => { + existing.consumers = Math.max(0, existing.consumers - 1); + if (existing.consumers === 0) { + existing.controller.abort(); + } + }; + req.signal.addEventListener('abort', onAbort, { once: true }); - // Save to cache - ttsAudioCache.set(cacheKey, buffer); + try { + const buffer = await existing.promise; + return new NextResponse(buffer, { + headers: { + 'Content-Type': contentType, + 'X-Cache': 'INFLIGHT', + 'ETag': etag, + 'Content-Length': String(buffer.byteLength), + 'Cache-Control': 'private, max-age=1800', + 'Vary': 'x-tts-provider, x-openai-key, x-openai-base-url' + } + }); + } finally { + try { req.signal.removeEventListener('abort', onAbort); } catch {} + } + } + + const controller = new AbortController(); + const entry: InflightEntry = { + controller, + consumers: 1, + promise: (async () => { + const release = await ttsSemaphore.acquire(); + try { + const buffer = await fetchTTSBufferWithRetry(openai, createParams, controller.signal); + // Save to cache + ttsAudioCache.set(cacheKey, buffer); + return buffer; + } finally { + release(); + inflightRequests.delete(cacheKey); + } + })() + }; + + inflightRequests.set(cacheKey, entry); + + const onAbort = (_evt: Event) => { + entry.consumers = Math.max(0, entry.consumers - 1); + if (entry.consumers === 0) { + entry.controller.abort(); + } + }; + req.signal.addEventListener('abort', onAbort, { once: true }); + + let buffer: ArrayBuffer; + try { + buffer = await entry.promise; + } finally { + try { req.signal.removeEventListener('abort', onAbort); } catch {} + } return new NextResponse(buffer, { headers: { 'Content-Type': contentType, - 'X-Cache': 'MISS' + 'X-Cache': 'MISS', + 'ETag': etag, + 'Content-Length': String(buffer.byteLength), + 'Cache-Control': 'private, max-age=1800', + 'Vary': 'x-tts-provider, x-openai-key, x-openai-base-url' } }); } catch (error) { diff --git a/tests/folders.spec.ts b/tests/folders.spec.ts index b435e41..268da6c 100644 --- a/tests/folders.spec.ts +++ b/tests/folders.spec.ts @@ -1,5 +1,5 @@ import { test, expect } from '@playwright/test'; -import { setupTest, uploadFiles, ensureDocumentsListed, deleteAllLocalDocuments } from './helpers'; +import { setupTest, uploadFiles, ensureDocumentsListed, deleteAllLocalDocuments, waitForDocumentListHintPersist } from './helpers'; test.describe('Document folders and hint persistence', () => { test.beforeEach(async ({ page }) => { @@ -88,6 +88,9 @@ test.describe('Document folders and hint persistence', () => { // Hint should disappear await expect(hint).toHaveCount(0); + // Ensure the dismissal has been persisted to IndexedDB before reloading + await waitForDocumentListHintPersist(page, false); + // Reload and ensure it remains dismissed await page.reload(); await page.waitForLoadState('networkidle'); diff --git a/tests/helpers.ts b/tests/helpers.ts index 240ce45..f7a8e25 100644 --- a/tests/helpers.ts +++ b/tests/helpers.ts @@ -49,20 +49,8 @@ export async function waitAndClickPlay(page: Page) { await expect(page.getByRole('button', { name: 'Play' })).toBeVisible(); // Play the TTS by clicking the button await page.getByRole('button', { name: 'Play' }).click(); - - // Expect for buttons to be disabled - await expect(page.locator('button[aria-label="Skip forward"][disabled]')).toBeVisible(); - await expect(page.locator('button[aria-label="Skip backward"][disabled]')).toBeVisible(); - - // Wait for the TTS to stop processing - await Promise.all([ - page.waitForSelector('button[aria-label="Skip forward"]:not([disabled])', { timeout: 45000 }), - page.waitForSelector('button[aria-label="Skip backward"]:not([disabled])', { timeout: 45000 }), - ]); - - await page.waitForFunction(() => { - return navigator.mediaSession?.playbackState === 'playing'; - }); + // Use resilient processing transition helper (tolerates fast completion) + await expectProcessingTransition(page); } /** @@ -93,7 +81,7 @@ export async function pauseTTSAndVerify(page: Page) { export async function setupTest(page: Page) { // Navigate to the home page before each test await page.goto('/'); - //await page.waitForLoadState('networkidle'); + await page.waitForLoadState('networkidle'); // If running in CI, select the "Custom OpenAI-Like" model and "Deepinfra" provider if (process.env.CI) { @@ -323,7 +311,37 @@ export async function changeNativeSpeedAndAssert(page: Page, newSpeed: number) { // Expect navigator.mediaSession.playbackState to equal given state export async function expectMediaState(page: Page, state: 'playing' | 'paused') { - await page.waitForFunction((s) => navigator.mediaSession?.playbackState === s, state, { timeout: 20000 }); + // WebKit (and sometimes other engines) may not reliably update navigator.mediaSession.playbackState. + // Fallback heuristics: + // 1. Prefer mediaSession if it matches desired state. + // 2. Otherwise inspect any