diff --git a/src/app/utils/mediaConcurrency.test.ts b/src/app/utils/mediaConcurrency.test.ts new file mode 100644 index 000000000..2d22f5bf8 --- /dev/null +++ b/src/app/utils/mediaConcurrency.test.ts @@ -0,0 +1,41 @@ +import { describe, expect, it } from 'vitest'; +import { withMediaFetchSlot } from './mediaConcurrency'; + +describe('withMediaFetchSlot', () => { + it('never runs more than three tasks at once and runs all of them', async () => { + let active = 0; + let peak = 0; + const release: (() => void)[] = []; + + const tasks = Array.from({ length: 10 }, () => + withMediaFetchSlot(async () => { + active += 1; + peak = Math.max(peak, active); + await new Promise((resolve) => { + release.push(resolve); + }); + active -= 1; + }) + ); + + while (release.length > 0) { + release.shift()?.(); + // Let the freed slot reach the next waiter before draining again. + // eslint-disable-next-line no-await-in-loop + await new Promise((resolve) => { + setTimeout(resolve, 0); + }); + } + + await Promise.all(tasks); + expect(peak).toBe(3); + }); + + it('releases the slot when a task rejects', async () => { + await expect(withMediaFetchSlot(() => Promise.reject(new Error('boom')))).rejects.toThrow( + 'boom' + ); + + await expect(withMediaFetchSlot(() => Promise.resolve('ok'))).resolves.toBe('ok'); + }); +}); diff --git a/src/app/utils/mediaConcurrency.ts b/src/app/utils/mediaConcurrency.ts new file mode 100644 index 000000000..142980126 --- /dev/null +++ b/src/app/utils/mediaConcurrency.ts @@ -0,0 +1,25 @@ +// Media shares its host with the Matrix API, and browsers cap connections per host at 6. +// Leave slots free so a wave of media loads cannot queue /sync and /send behind it. +const MAX_CONCURRENT_MEDIA_FETCHES = 3; + +const waiters: (() => void)[] = []; +let active = 0; + +export async function withMediaFetchSlot(task: () => Promise): Promise { + if (active < MAX_CONCURRENT_MEDIA_FETCHES) { + active += 1; + } else { + await new Promise((resolve) => { + waiters.push(resolve); + }); + } + + try { + return await task(); + } finally { + // Hand the slot straight to the next waiter instead of releasing and re-acquiring it. + const next = waiters.shift(); + if (next) next(); + else active -= 1; + } +} diff --git a/src/app/utils/mediaTransport.ts b/src/app/utils/mediaTransport.ts index 34083eb76..186c0b2ad 100644 --- a/src/app/utils/mediaTransport.ts +++ b/src/app/utils/mediaTransport.ts @@ -1,6 +1,7 @@ import { getCachedSWMediaAuthSupport } from './swMediaAuth'; import { fetch } from '$utils/fetch'; import { getFromMediaCache, putInMediaCache } from './mediaCache'; +import { withMediaFetchSlot } from './mediaConcurrency'; type StoredSession = { baseUrl?: string; @@ -326,7 +327,9 @@ async function fetchMediaBlobInternal(url: string, options?: MediaTransportOptio }; if (useServiceWorker) { - return fetchAndCache(await fetchMediaResponse(url, undefined, cacheMode)); + return withMediaFetchSlot(async () => + fetchAndCache(await fetchMediaResponse(url, undefined, cacheMode)) + ); } const fetchWithRetry = async ( @@ -352,20 +355,22 @@ async function fetchMediaBlobInternal(url: string, options?: MediaTransportOptio } }; - const initialAccessToken = resolveAccessToken(url, options); - const initialResponse = await fetchWithRetry(url, initialAccessToken, cacheMode); - if (initialResponse.ok) { - return fetchAndCache(initialResponse); - } + return withMediaFetchSlot(async () => { + const initialAccessToken = resolveAccessToken(url, options); + const initialResponse = await fetchWithRetry(url, initialAccessToken, cacheMode); + if (initialResponse.ok) { + return fetchAndCache(initialResponse); + } - if (!isRetryableAuthError(initialResponse)) { - throw new Error( - `Failed to fetch media: ${initialResponse.status} ${initialResponse.statusText}` - ); - } + if (!isRetryableAuthError(initialResponse)) { + throw new Error( + `Failed to fetch media: ${initialResponse.status} ${initialResponse.statusText}` + ); + } - const retryAccessToken = resolveAccessToken(url, options); - return fetchAndCache(await fetchMediaResponse(url, retryAccessToken, cacheMode)); + const retryAccessToken = resolveAccessToken(url, options); + return fetchAndCache(await fetchMediaResponse(url, retryAccessToken, cacheMode)); + }); } export async function fetchMediaBlob(url: string, options?: MediaTransportOptions): Promise { diff --git a/src/sw.ts b/src/sw.ts index 5565b05f3..96093fcc4 100644 --- a/src/sw.ts +++ b/src/sw.ts @@ -4,6 +4,7 @@ import { precacheAndRoute, cleanupOutdatedCaches } from 'workbox-precaching'; import { createPushNotifications } from './sw/pushNotification'; +import { withMediaFetchSlot } from './app/utils/mediaConcurrency'; declare const self: ServiceWorkerGlobalScope; @@ -790,8 +791,9 @@ function respondWithInflightMedia( // Fetch by URL instead of reusing the subresource Request. Image requests commonly carry // mode: "no-cors", which prevents the Authorization header above from reaching the server. // Preserve Range header for streaming audio and video. - const promise = fetch(request.url, { ...fetchConfig(token, request), redirect }) - .then( + // The slot is held until the body has been read, since that is what holds the connection. + const promise = withMediaFetchSlot(() => + fetch(request.url, { ...fetchConfig(token, request), redirect }).then( async (res): Promise => ({ status: res.status, statusText: res.statusText, @@ -799,9 +801,9 @@ function respondWithInflightMedia( body: await res.arrayBuffer(), }) ) - .finally(() => { - inflightMediaFetches.delete(key); - }); + ).finally(() => { + inflightMediaFetches.delete(key); + }); inflightMediaFetches.set(key, promise); return promise.then( (data) =>