From ab19c0d1ac2d2f4930160d553b4d89cb0cb74b2c Mon Sep 17 00:00:00 2001 From: Hampus Date: Mon, 12 Oct 2026 15:39:30 +0200 Subject: [PATCH] fix(desktop): never queue local proxy requests (#3421) --- .../src/main/DesktopOutboundHTTP.test.mjs | 55 ---- .../src/main/DesktopOutboundHTTP.ts | 58 +--- .../src/main/DesktopRuntimeSecurity.test.mjs | 12 +- .../src/main/DesktopRuntimeSecurity.ts | 15 +- .../main/DesktopTrustedCertificates.test.mjs | 1 - .../src/main/LocalAppProxyClient.ts | 179 +----------- .../main/LocalAppProxyConcurrency.test.mjs | 269 ++++-------------- .../src/main/SelectedInstanceFetch.ts | 2 - 8 files changed, 86 insertions(+), 505 deletions(-) diff --git a/fluxer_desktop/src/main/DesktopOutboundHTTP.test.mjs b/fluxer_desktop/src/main/DesktopOutboundHTTP.test.mjs index 1e8bfbc93..65b7a02f2 100644 --- a/fluxer_desktop/src/main/DesktopOutboundHTTP.test.mjs +++ b/fluxer_desktop/src/main/DesktopOutboundHTTP.test.mjs @@ -65,7 +65,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { headers: null, method: 'GET', originTrust: DesktopOriginTrust.REGISTERED, - serviceName: 'test', signal: null, timeoutMs: 500, url: 'https://media.internal/avatars/1.webp', @@ -104,7 +103,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { headers: null, method: 'GET', originTrust: DesktopOriginTrust.REGISTERED, - serviceName: 'test', signal: null, timeoutMs: 500, url: 'https://media.internal/avatars/1.webp', @@ -161,7 +159,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { headers: null, method: 'GET', originTrust: DesktopOriginTrust.REGISTERED, - serviceName: 'test', signal: null, timeoutMs: 500, url: 'https://media.mixed.example/avatars/1.webp', @@ -193,44 +190,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { outboundHTTP.cleanup(); }); - test('a request that cannot be constructed releases its in-flight slot', async () => { - const {port, server} = await startServer({ - '/ok': (_request, response) => sendImage(response, 'image/png', PNG_BYTES), - }); - try { - const outboundHTTP = new DesktopOutboundHTTP({ - resolveHostAddresses: stubResolver({'lan.example': ['127.0.0.1']}), - }); - await outboundHTTP.registerAnchoredOrigins({ - anchorOrigin: `http://lan.example:${port}`, - origins: [`http://lan.example:${port}`], - }); - const send = (headers) => - outboundHTTP.request({ - body: null, - expectedOrigin: `http://lan.example:${port}`, - headers, - method: 'GET', - originTrust: DesktopOriginTrust.REGISTERED, - serviceName: 'slot-leak', - signal: null, - timeoutMs: 2000, - url: `http://lan.example:${port}/ok`, - }); - - for (let attempt = 0; attempt < 80; attempt += 1) { - await assert.rejects(send({'x-bad': 'a\u0001b'})); - } - const message = await send(null); - - assert.equal(message.status, 200); - message.message.resume(); - outboundHTTP.cleanup(); - } finally { - server.close(); - } - }); - test('a request whose signal already aborted is refused instead of sent', async () => { const seen = []; const {port, server} = await startServer({ @@ -257,7 +216,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { headers: null, method: 'GET', originTrust: DesktopOriginTrust.REGISTERED, - serviceName: 'aborted', signal: controller.signal, timeoutMs: 2000, url: `http://lan.example:${port}/ok`, @@ -289,7 +247,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { headers: null, method: 'GET', originTrust: DesktopOriginTrust.REGISTERED, - serviceName: 'test', signal: null, timeoutMs: 2000, url: `http://media.lan.example:${port}/media/ok`, @@ -331,7 +288,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { headers: {'content-type': 'application/json', 'x-fluxer-sudo-mode-jwt': 'previous'}, method, originTrust: DesktopOriginTrust.REGISTERED, - serviceName: 'test', signal: null, timeoutMs: 2000, url: `http://lan.example:${port}/api/v1/oauth2/applications/1`, @@ -549,7 +505,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { headers: null, method: 'GET', originTrust: DesktopOriginTrust.BOUND, - serviceName: 'test', signal, timeoutMs: 2000, url: `https://home.example:${port}/ping`, @@ -577,7 +532,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { headers: null, method: 'GET', originTrust: DesktopOriginTrust.BOUND, - serviceName: 'test', signal: null, timeoutMs: 200, url: 'http://blackhole.example:9/ping', @@ -606,7 +560,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { headers: null, method: 'GET', originTrust: DesktopOriginTrust.BOUND, - serviceName: 'test', signal: null, timeoutMs: 2000, url: `http://mixed.example:${port}/ping`, @@ -632,7 +585,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { headers: null, method: 'GET', originTrust: DesktopOriginTrust.BOUND, - serviceName: 'test', signal: null, timeoutMs: 2000, url: `http://127.0.0.1:${other.port}/steal`, @@ -670,7 +622,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { headers: null, method: 'GET', originTrust: DesktopOriginTrust.BOUND, - serviceName: 'test', signal: null, timeoutMs: 5000, url: `http://127.0.0.1:${port}/ping`, @@ -720,7 +671,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { headers: {'content-type': 'text/plain'}, method: 'PUT', originTrust: DesktopOriginTrust.BOUND, - serviceName: 'test', signal: null, timeoutMs: 5000, url: `http://127.0.0.1:${port}/upload`, @@ -775,7 +725,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { headers: null, method: 'GET', originTrust: DesktopOriginTrust.REGISTERED, - serviceName: 'test', signal: null, timeoutMs: 2000, url: 'https://unrouted.example/', @@ -826,7 +775,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { headers: null, method: 'GET', originTrust: DesktopOriginTrust.REGISTERED, - serviceName: 'test', signal: null, timeoutMs: 2000, url: `http://dual.example:${port}/ping`, @@ -857,7 +805,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => { headers: null, method: 'GET', originTrust: DesktopOriginTrust.REGISTERED, - serviceName: 'test', signal: null, timeoutMs: 3000, url: `http://multi.example:${port}/ping`, @@ -1007,7 +954,6 @@ describe('DesktopOutboundHTTP system proxy routes', () => { headers: {Accept: 'image/png'}, method: 'GET', originTrust: DesktopOriginTrust.REGISTERED, - serviceName: 'test', signal: null, timeoutMs: 500, url: 'https://media.blocked.example/avatar.png', @@ -1061,7 +1007,6 @@ describe('DesktopOutboundHTTP system proxy routes', () => { headers: null, method: 'GET', originTrust, - serviceName: 'test', signal: null, timeoutMs: 500, url, diff --git a/fluxer_desktop/src/main/DesktopOutboundHTTP.ts b/fluxer_desktop/src/main/DesktopOutboundHTTP.ts index dded05400..bd7698d9f 100644 --- a/fluxer_desktop/src/main/DesktopOutboundHTTP.ts +++ b/fluxer_desktop/src/main/DesktopOutboundHTTP.ts @@ -28,8 +28,6 @@ const DESKTOP_OUTBOUND_HTTP_RESOLUTION_TIMEOUT_MS = 10_000; const DESKTOP_OUTBOUND_HTTP_MAX_TARGET_URL_BYTES = 16 * 1024; const DESKTOP_OUTBOUND_HTTP_MAX_REDIRECT_LOCATION_BYTES = 4096; const DESKTOP_OUTBOUND_HTTP_MAX_ORIGINS = 2048; -const DESKTOP_OUTBOUND_HTTP_MAX_ORIGIN_REQUESTS_IN_FLIGHT = 128; -const DESKTOP_OUTBOUND_HTTP_MAX_IN_FLIGHT_PER_SERVICE = 64; const DESKTOP_OUTBOUND_HTTP_MAX_PENDING_RESOLUTIONS = 64; const DESKTOP_OUTBOUND_HTTP_MAX_REQUEST_BODY_BYTES = 1024 * 1024; const DESKTOP_OUTBOUND_HTTP_STALE_SOCKET_ATTEMPTS = 1; @@ -113,7 +111,6 @@ export interface DesktopOutboundHTTPRequest { readonly maximumRequestBodyBytes?: number; readonly method: string; readonly originTrust: DesktopOriginTrust; - readonly serviceName: string; readonly signal: AbortSignal | null; readonly timeoutMs: number; readonly url: string; @@ -565,7 +562,7 @@ interface DesktopProxiedSend { readonly body: Uint8Array | Readable | null; readonly headers: Readonly>; readonly method: string; - readonly release: () => void; + readonly release?: () => void; readonly signal: AbortSignal | null; readonly target: URL; readonly timeout: () => Error; @@ -578,20 +575,15 @@ export class DesktopOutboundHTTP { maxSockets: DESKTOP_OUTBOUND_HTTP_MAX_SOCKETS, }); private readonly httpsAgent: https.Agent; - private readonly originRequestHttpAgent = new HappyEyeballsHttpAgent({ - keepAlive: true, - maxSockets: DESKTOP_OUTBOUND_HTTP_MAX_IN_FLIGHT_PER_SERVICE, - }); + private readonly originRequestHttpAgent = new HappyEyeballsHttpAgent({keepAlive: true}); private readonly originRequestHttpsAgent: https.Agent; private readonly bindings = new Map(); private readonly pendingBindings = new Map(); private readonly registeredRequirements = new Map(); - private readonly originRequestsInFlightByService = new Map(); private readonly resolveHostAddresses: DesktopHostAddressResolver; private readonly resolveProxy: DesktopProxyResolver; private readonly sendThroughSession: DesktopSessionHTTPSender; private inFlight = 0; - private originRequestsInFlight = 0; private activeResolutions = 0; private acceptingRequests = true; @@ -608,7 +600,6 @@ export class DesktopOutboundHTTP { }); this.originRequestHttpsAgent = new HappyEyeballsHttpsAgent({ keepAlive: true, - maxSockets: DESKTOP_OUTBOUND_HTTP_MAX_IN_FLIGHT_PER_SERVICE, ...trust, }); } @@ -783,7 +774,6 @@ export class DesktopOutboundHTTP { this.originRequestHttpsAgent.destroy(); this.bindings.clear(); this.registeredRequirements.clear(); - this.originRequestsInFlightByService.clear(); } private async routesDirect(url: URL): Promise { @@ -813,15 +803,6 @@ export class DesktopOutboundHTTP { request: DesktopOutboundHTTPRequest, target: URL, ): Promise { - this.acquireSlot(request.serviceName); - let released = false; - const release = (): void => { - if (released) { - return; - } - released = true; - this.releaseSlot(request.serviceName); - }; return this.sendProxied({ body: this.proxiedRequestBody(request), headers: requestHeaders(request.headers), @@ -830,7 +811,6 @@ export class DesktopOutboundHTTP { target, timeout: () => new DesktopOutboundHTTPRequestTimeoutError(target.toString(), request.timeoutMs), timeoutMs: request.timeoutMs, - release, }).catch((error: unknown) => { if (request.signal?.aborted === true) { throw new DesktopOutboundHTTPRequestAbortedError(target.toString()); @@ -854,7 +834,7 @@ export class DesktopOutboundHTTP { timer.unref(); const settle = (): void => { clearTimeout(timer); - release(); + release?.(); }; try { const response = await this.sendThroughSession({ @@ -1083,42 +1063,11 @@ export class DesktopOutboundHTTP { } } - private acquireSlot(serviceName: string): void { - if (this.originRequestsInFlight >= DESKTOP_OUTBOUND_HTTP_MAX_ORIGIN_REQUESTS_IN_FLIGHT) { - throw new DesktopOutboundHTTPCapacityError('in-flight request'); - } - const perService = this.originRequestsInFlightByService.get(serviceName) ?? 0; - if (perService >= DESKTOP_OUTBOUND_HTTP_MAX_IN_FLIGHT_PER_SERVICE) { - throw new DesktopOutboundHTTPCapacityError(`in-flight request for ${serviceName}`); - } - this.originRequestsInFlight += 1; - this.originRequestsInFlightByService.set(serviceName, perService + 1); - } - - private releaseSlot(serviceName: string): void { - this.originRequestsInFlight -= 1; - const perService = (this.originRequestsInFlightByService.get(serviceName) ?? 1) - 1; - if (perService <= 0) { - this.originRequestsInFlightByService.delete(serviceName); - } else { - this.originRequestsInFlightByService.set(serviceName, perService); - } - } - private issueOriginRequest( request: DesktopOutboundHTTPRequest, target: URL, binding: DesktopOriginAddressBinding, ): Promise { - this.acquireSlot(request.serviceName); - let released = false; - const release = (): void => { - if (released) { - return; - } - released = true; - this.releaseSlot(request.serviceName); - }; return new Promise((resolve, reject) => { let attemptsRemaining = isReplayableRequest(request) ? DESKTOP_OUTBOUND_HTTP_STALE_SOCKET_ATTEMPTS : 0; let active: http.ClientRequest | null = null; @@ -1144,7 +1093,6 @@ export class DesktopOutboundHTTP { const settle = (): void => { clearTimeout(timeout); request.signal?.removeEventListener('abort', onAbort); - release(); }; const destroyActive = (error: Error): void => { active?.destroy(error); diff --git a/fluxer_desktop/src/main/DesktopRuntimeSecurity.test.mjs b/fluxer_desktop/src/main/DesktopRuntimeSecurity.test.mjs index 9e1f54423..f99df1b6a 100644 --- a/fluxer_desktop/src/main/DesktopRuntimeSecurity.test.mjs +++ b/fluxer_desktop/src/main/DesktopRuntimeSecurity.test.mjs @@ -16,10 +16,12 @@ function createLogger() { function createSession() { const session = { + requestFilter: null, requestListener: null, webRequest: { - onBeforeSendHeaders: (listener) => { - session.requestListener = listener; + onBeforeSendHeaders: (...args) => { + session.requestFilter = args.length > 1 ? args[0] : null; + session.requestListener = args.at(-1); }, }, }; @@ -141,6 +143,12 @@ describe('DesktopRuntimeSecurity', () => { ); }); + test('only intercepts local app requests and the YouTube embed frame', () => { + const {session} = createInstalled(); + + assert.deepEqual(session.requestFilter, {urls: ['fluxer-app://*/*', 'https://www.youtube.com/embed/*']}); + }); + test('installs once per session and stops listening after cleanup', () => { const {security, session} = createInstalled(); const firstRequestListener = session.requestListener; diff --git a/fluxer_desktop/src/main/DesktopRuntimeSecurity.ts b/fluxer_desktop/src/main/DesktopRuntimeSecurity.ts index 379ec596a..158e1c35f 100644 --- a/fluxer_desktop/src/main/DesktopRuntimeSecurity.ts +++ b/fluxer_desktop/src/main/DesktopRuntimeSecurity.ts @@ -1,6 +1,6 @@ // SPDX-License-Identifier: AGPL-3.0-or-later -import {STABLE_APP_URL} from '@electron/common/Constants'; +import {DESKTOP_APP_SCHEME, STABLE_APP_URL} from '@electron/common/Constants'; import {isPlainRecord} from '@electron/common/PlainRecord'; const SUB_FRAME_RESOURCE_TYPE = 'subFrame'; @@ -9,6 +9,10 @@ const REFERER_REQUEST_HEADER = 'Referer'; const REFERER_HEADER_NAME_LOWERCASE = 'referer'; const YOUTUBE_EMBED_FRAME_ORIGIN = 'https://www.youtube.com'; const YOUTUBE_EMBED_FRAME_PATH_PREFIX = '/embed/'; +const DESKTOP_RUNTIME_SECURITY_REQUEST_URL_PATTERNS = [ + `${DESKTOP_APP_SCHEME}://*/*`, + `${YOUTUBE_EMBED_FRAME_ORIGIN}${YOUTUBE_EMBED_FRAME_PATH_PREFIX}*`, +]; interface DesktopRuntimeSecurityLogger { error: (...args: Array) => void; @@ -62,9 +66,12 @@ export class DesktopRuntimeSecurity { } this.installedSessions.add(session); try { - session.webRequest.onBeforeSendHeaders((details: unknown, callback: unknown) => { - this.settleRequestHeaders(details, callback); - }); + session.webRequest.onBeforeSendHeaders( + {urls: DESKTOP_RUNTIME_SECURITY_REQUEST_URL_PATTERNS}, + (details: unknown, callback: unknown) => { + this.settleRequestHeaders(details, callback); + }, + ); } catch (installationError) { this.uninstall(session); throw installationError; diff --git a/fluxer_desktop/src/main/DesktopTrustedCertificates.test.mjs b/fluxer_desktop/src/main/DesktopTrustedCertificates.test.mjs index aefa9914b..051b87bc2 100644 --- a/fluxer_desktop/src/main/DesktopTrustedCertificates.test.mjs +++ b/fluxer_desktop/src/main/DesktopTrustedCertificates.test.mjs @@ -41,7 +41,6 @@ async function fetchThrough(certificates, port) { headers: null, method: 'GET', originTrust: DesktopOriginTrust.REGISTERED, - serviceName: 'test', signal: null, timeoutMs: 3000, url: `https://ca.example:${port}/`, diff --git a/fluxer_desktop/src/main/LocalAppProxyClient.ts b/fluxer_desktop/src/main/LocalAppProxyClient.ts index 4c7f78ba1..df40fc9d8 100644 --- a/fluxer_desktop/src/main/LocalAppProxyClient.ts +++ b/fluxer_desktop/src/main/LocalAppProxyClient.ts @@ -20,8 +20,6 @@ const LOCAL_APP_PROXY_REQUEST_BODY_MAX_BYTES = 100 * 1024 * 1024; export const LOCAL_APP_PROXY_ACCEPT_ENCODING = 'gzip, br'; const LOCAL_APP_PROXY_RESOURCE_RESPONSE_HEADERS_DEADLINE_MS = 30_000; -const LOCAL_APP_PROXY_MAX_CONCURRENT_REQUESTS = 32; -const LOCAL_APP_PROXY_MAX_QUEUED_REQUESTS = 4096; const UPLOAD_PROGRESS_MIN_BYTES = 64 * 1024; const UPLOAD_PROGRESS_MIN_INTERVAL_MS = 100; const MINIMUM_RESPONSE_STATUS = 200; @@ -42,101 +40,6 @@ class LocalAppProxyResponseHeadersDeadlineError extends Error { } } -class LocalAppProxyQueueOverflowError extends Error { - public constructor() { - super('Local app proxy has too many queued requests'); - this.name = 'LocalAppProxyQueueOverflowError'; - } -} - -class LocalAppProxyConcurrencyPermit { - private releasePermit: (() => void) | null; - - public constructor(releasePermit: () => void) { - this.releasePermit = releasePermit; - } - - public release(): void { - const releasePermit = this.releasePermit; - if (releasePermit == null) { - return; - } - this.releasePermit = null; - releasePermit(); - } -} - -class LocalAppProxyResponseBodyLease { - public readonly reader: ReadableStreamDefaultReader; - private readonly permit: LocalAppProxyConcurrencyPermit; - private settled = false; - - public constructor(body: ReadableStream, permit: LocalAppProxyConcurrencyPermit) { - this.reader = body.getReader(); - this.permit = permit; - } - - public release(): void { - if (this.settled) { - return; - } - this.settled = true; - try { - this.reader.releaseLock(); - } finally { - this.permit.release(); - } - } -} - -class LocalAppProxyConcurrencyGate { - private active = 0; - private readonly waiting: Array<{readonly admit: () => void}> = []; - private readonly newestFirst: boolean; - - public constructor(newestFirst: boolean) { - this.newestFirst = newestFirst; - } - - public acquire(signal: AbortSignal): Promise { - signal.throwIfAborted(); - if (this.active < LOCAL_APP_PROXY_MAX_CONCURRENT_REQUESTS) { - this.active += 1; - return Promise.resolve(new LocalAppProxyConcurrencyPermit(() => this.release())); - } - if (this.waiting.length >= LOCAL_APP_PROXY_MAX_QUEUED_REQUESTS) { - throw new LocalAppProxyQueueOverflowError(); - } - return new Promise((resolve, reject) => { - let waiter: (typeof this.waiting)[number]; - const onAbort = () => { - const index = this.waiting.indexOf(waiter); - if (index === -1) { - return; - } - this.waiting.splice(index, 1); - reject(signal.reason); - }; - const admit = () => { - signal.removeEventListener('abort', onAbort); - resolve(new LocalAppProxyConcurrencyPermit(() => this.release())); - }; - waiter = {admit}; - this.waiting.push(waiter); - signal.addEventListener('abort', onAbort, {once: true}); - }); - } - - private release(): void { - const next = this.newestFirst ? this.waiting.pop() : this.waiting.shift(); - if (next != null) { - next.admit(); - return; - } - this.active -= 1; - } -} - interface DesktopLocalAppProxyClientDependencies { readonly outboundHTTP: DesktopOutboundHTTP; readonly traffic: LocalAppProxyTraffic; @@ -226,35 +129,20 @@ function emitLocalAppUploadProgress(progress: DesktopLocalAppUploadProgress): vo export class DesktopLocalAppProxyClient { private readonly outboundHTTP: DesktopOutboundHTTP; - private readonly traffic: LocalAppProxyTraffic; - private readonly concurrency: LocalAppProxyConcurrencyGate; private readonly responseHeadersDeadlineMs: number | null; public constructor(dependencies: DesktopLocalAppProxyClientDependencies) { this.outboundHTTP = dependencies.outboundHTTP; - this.traffic = dependencies.traffic; - const remoteResource = dependencies.traffic === LocalAppProxyTraffic.REMOTE_RESOURCE; - this.concurrency = new LocalAppProxyConcurrencyGate(remoteResource); - this.responseHeadersDeadlineMs = remoteResource ? LOCAL_APP_PROXY_RESOURCE_RESPONSE_HEADERS_DEADLINE_MS : null; + this.responseHeadersDeadlineMs = + dependencies.traffic === LocalAppProxyTraffic.REMOTE_RESOURCE + ? LOCAL_APP_PROXY_RESOURCE_RESPONSE_HEADERS_DEADLINE_MS + : null; } public async fetch(request: LocalAppProxyFetchRequest): Promise { - const permit = await this.concurrency.acquire(request.signal); - try { - return await this.runFetchWithinHeadersDeadline(request, permit); - } catch (error) { - permit.release(); - throw error; - } - } - - private async runFetchWithinHeadersDeadline( - request: LocalAppProxyFetchRequest, - permit: LocalAppProxyConcurrencyPermit, - ): Promise { const deadlineMs = this.responseHeadersDeadlineMs; if (deadlineMs == null || requestHasBody(request.method)) { - return await this.runFetch(request, permit); + return await this.runFetch(request); } const deadline = new AbortController(); const timer = setTimeout( @@ -263,16 +151,13 @@ export class DesktopLocalAppProxyClient { ); timer.unref(); try { - return await this.runFetch({...request, signal: AbortSignal.any([request.signal, deadline.signal])}, permit); + return await this.runFetch({...request, signal: AbortSignal.any([request.signal, deadline.signal])}); } finally { clearTimeout(timer); } } - private async runFetch( - request: LocalAppProxyFetchRequest, - permit: LocalAppProxyConcurrencyPermit, - ): Promise { + private async runFetch(request: LocalAppProxyFetchRequest): Promise { const target = new URL(request.targetURL); const body = requestHasBody(request.method) ? request.body : null; const outboundRequest: DesktopOutboundHTTPRequest = { @@ -282,12 +167,11 @@ export class DesktopLocalAppProxyClient { maximumRequestBodyBytes: LOCAL_APP_PROXY_REQUEST_BODY_MAX_BYTES, method: request.method, originTrust: DesktopOriginTrust.REGISTERED, - serviceName: this.traffic, signal: request.signal, timeoutMs: LOCAL_APP_PROXY_DEADLINE_MS, url: target.href, }; - return buildProxyResponse(await this.outboundHTTP.request(outboundRequest), request.method, permit); + return buildProxyResponse(await this.outboundHTTP.request(outboundRequest), request.method); } public async settleFailure({ @@ -368,11 +252,7 @@ function instrumentedRequestBody( ); } -function buildProxyResponse( - message: DesktopOutboundHTTPMessage, - method: string, - permit: LocalAppProxyConcurrencyPermit, -): Response { +function buildProxyResponse(message: DesktopOutboundHTTPMessage, method: string): Response { try { const status = message.status; if (status < MINIMUM_RESPONSE_STATUS || status > MAXIMUM_RESPONSE_STATUS) { @@ -381,59 +261,20 @@ function buildProxyResponse( const headers = responseHeaders(message.headers); if (method === LOCAL_APP_PROXY_HEAD_METHOD || NULL_BODY_STATUSES.has(status)) { message.message.resume(); - permit.release(); return new Response(null, {status, statusText: message.statusText, headers}); } const decoded = decodedMessageStream(message.message, contentEncoding(message.headers)); - const body = holdPermitForResponseBody(Readable.toWeb(decoded) as ReadableStream, permit); - return new Response(body, { + return new Response(Readable.toWeb(decoded) as ReadableStream, { status, statusText: message.statusText, headers, }); } catch (error) { message.message.destroy(); - permit.release(); throw error; } } -function holdPermitForResponseBody( - body: ReadableStream, - permit: LocalAppProxyConcurrencyPermit, -): ReadableStream { - const lease = new LocalAppProxyResponseBodyLease(body, permit); - return new ReadableStream({ - async pull(controller) { - try { - const chunk = await lease.reader.read(); - if (chunk.done) { - try { - controller.close(); - } finally { - lease.release(); - } - return; - } - controller.enqueue(chunk.value); - } catch (error) { - try { - controller.error(error); - } finally { - lease.release(); - } - } - }, - async cancel(reason) { - try { - await lease.reader.cancel(reason); - } finally { - lease.release(); - } - }, - }); -} - function responseHeaders(incoming: http.IncomingHttpHeaders): Headers { const headers = new Headers(); for (const [name, value] of Object.entries(incoming)) { diff --git a/fluxer_desktop/src/main/LocalAppProxyConcurrency.test.mjs b/fluxer_desktop/src/main/LocalAppProxyConcurrency.test.mjs index e9871273b..d7e97094d 100644 --- a/fluxer_desktop/src/main/LocalAppProxyConcurrency.test.mjs +++ b/fluxer_desktop/src/main/LocalAppProxyConcurrency.test.mjs @@ -2,7 +2,7 @@ import assert from 'node:assert/strict'; import {Buffer} from 'node:buffer'; -import {Readable} from 'node:stream'; +import {PassThrough, Readable} from 'node:stream'; import {describe, mock, test} from 'node:test'; import {installElectronStub} from './LocalAppTestSupport.test.mjs'; @@ -10,17 +10,9 @@ installElectronStub(); const {DesktopLocalAppProxyClient, LocalAppProxyTraffic} = await import('./LocalAppProxyClient.ts'); -const PROXY_MAX_CONCURRENT_REQUESTS = 32; -const PROXY_MAX_QUEUED_REQUESTS = 4096; -const OUTBOUND_MAX_IN_FLIGHT_PER_SERVICE = 64; const RENDERED_AVATAR_COUNT = 120; - -class DesktopOutboundHTTPCapacityError extends Error { - constructor(limit) { - super(`Desktop outbound HTTP exceeded its ${limit} limit`); - this.name = 'DesktopOutboundHTTPCapacityError'; - } -} +const PARKED_MEDIA_BODY_COUNT = 200; +const RESOURCE_HEADERS_DEADLINE_MS = 30_000; function tick() { return new Promise((resolve) => { @@ -28,33 +20,6 @@ function tick() { }); } -function createOutboundHTTP() { - const state = {inFlight: 0, peakInFlight: 0, started: 0, pending: []}; - const outboundHTTP = { - request: (request) => { - if (state.inFlight >= OUTBOUND_MAX_IN_FLIGHT_PER_SERVICE) { - throw new DesktopOutboundHTTPCapacityError(`in-flight request for ${request.serviceName}`); - } - state.inFlight += 1; - state.started += 1; - state.peakInFlight = Math.max(state.peakInFlight, state.inFlight); - return new Promise((resolve) => { - state.pending.push(() => { - state.inFlight -= 1; - resolve({ - headers: {'content-type': 'image/webp'}, - message: Readable.from([Buffer.from('avatar-bytes')]), - status: 200, - statusText: 'OK', - url: new URL(request.url), - }); - }); - }); - }, - }; - return {outboundHTTP, state}; -} - function avatarRequest(index, signal) { return { targetURL: `https://media.fluxer.app/avatars/${index}.webp`, @@ -68,131 +33,17 @@ function avatarRequest(index, signal) { }; } -async function pump(state, work) { - let finished = false; - const result = work.finally(() => { - finished = true; - }); - result.catch(() => {}); - while (!finished) { - for (const settle of state.pending.splice(0)) { - settle(); - } - await tick(); - } - return await result; -} - -function readAvatar(client, index, signal) { - return client.fetch(avatarRequest(index, signal)).then(async (response) => ({ - status: response.status, - text: await response.text(), - })); -} - -describe('the local app proxy bounds its own concurrency instead of overrunning outbound HTTP', () => { - test('every avatar of a fully rendered chat resolves rather than tripping the outbound capacity cap', async () => { - const {outboundHTTP, state} = createOutboundHTTP(); - const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE}); - const controller = new AbortController(); - - const fetches = []; - for (let index = 0; index < RENDERED_AVATAR_COUNT; index += 1) { - fetches.push(readAvatar(client, index, controller.signal)); - } - const responses = await pump(state, Promise.all(fetches)); - - assert.equal(responses.length, RENDERED_AVATAR_COUNT); - assert.deepEqual(new Set(responses.map((response) => response.status)), new Set([200])); - assert.equal(state.started, RENDERED_AVATAR_COUNT); - assert.deepEqual( - responses.map((response) => response.text), - new Array(RENDERED_AVATAR_COUNT).fill('avatar-bytes'), - ); - }); - - test('no more requests are in flight at any instant than the proxy concurrency limit', async () => { - const {outboundHTTP, state} = createOutboundHTTP(); - const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE}); - const controller = new AbortController(); - - const fetches = []; - for (let index = 0; index < RENDERED_AVATAR_COUNT; index += 1) { - fetches.push(readAvatar(client, index, controller.signal)); - } - await tick(); - const peakBeforeAnyCompletion = state.peakInFlight; - await pump(state, Promise.all(fetches)); - - assert.equal(peakBeforeAnyCompletion, PROXY_MAX_CONCURRENT_REQUESTS); - assert.equal(state.peakInFlight, PROXY_MAX_CONCURRENT_REQUESTS); - }); - - test('an aborted waiter rejects with the abort reason and leaves its queue slot to the next waiter', async () => { - const {outboundHTTP, state} = createOutboundHTTP(); - const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE}); - const held = new AbortController(); - - const holding = []; - for (let index = 0; index < PROXY_MAX_CONCURRENT_REQUESTS; index += 1) { - holding.push(readAvatar(client, index, held.signal)); - } - await tick(); - assert.equal(state.started, PROXY_MAX_CONCURRENT_REQUESTS); - - const abandoned = new AbortController(); - const abandonedFetch = client.fetch(avatarRequest(998, abandoned.signal)); - const survivor = readAvatar(client, 999, held.signal); - const reason = new Error('the renderer navigated away'); - abandoned.abort(reason); - - await assert.rejects(abandonedFetch, (error) => error === reason); - assert.equal(state.started, PROXY_MAX_CONCURRENT_REQUESTS); - - state.pending.shift()(); - await tick(); - await tick(); - assert.equal(state.started, PROXY_MAX_CONCURRENT_REQUESTS + 1); - - const settled = await pump(state, Promise.all([survivor, ...holding])); - assert.equal(settled[0].status, 200); - assert.equal(settled.length, PROXY_MAX_CONCURRENT_REQUESTS + 1); - }); - - test('the queue refuses a request past its bound rather than growing without limit', async () => { - const {outboundHTTP, state} = createOutboundHTTP(); - const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE}); - const controller = new AbortController(); - - const accepted = []; - for (let index = 0; index < PROXY_MAX_CONCURRENT_REQUESTS + PROXY_MAX_QUEUED_REQUESTS; index += 1) { - accepted.push(readAvatar(client, index, controller.signal)); - } - - await assert.rejects( - client.fetch(avatarRequest(Number.MAX_SAFE_INTEGER, controller.signal)), - (error) => error.name === 'LocalAppProxyQueueOverflowError', - ); - - const settled = await pump(state, Promise.all(accepted)); - assert.equal(settled.length, PROXY_MAX_CONCURRENT_REQUESTS + PROXY_MAX_QUEUED_REQUESTS); - }); -}); - -const RESOURCE_HEADERS_DEADLINE_MS = 30_000; - -function createSilentUpstream() { +function createUpstream({streamBodies}) { const state = {started: [], aborted: 0}; const outboundHTTP = { request: (request) => new Promise((resolve, reject) => { const entry = { url: request.url, - serviceName: request.serviceName, answer: () => resolve({ headers: {'content-type': 'image/webp'}, - message: Readable.from([Buffer.from('avatar-bytes')]), + message: streamBodies ? new PassThrough() : Readable.from([Buffer.from('avatar-bytes')]), status: 200, statusText: 'OK', url: new URL(request.url), @@ -212,80 +63,64 @@ function createSilentUpstream() { return {outboundHTTP, state}; } -function apiRequest(signal) { - return {...avatarRequest(0, signal), targetURL: 'https://api.fluxer.app/v1/users/@me'}; -} +describe('the local app proxy never queues a request behind another', () => { + test('every avatar of a fully rendered chat starts at once', async () => { + const {outboundHTTP, state} = createUpstream({streamBodies: false}); + const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE}); + const renderer = new AbortController(); -describe('media that a slow or silent self-host never answers cannot wedge the local app proxy', () => { - test('avatars dropped while their upstream sends no headers free their permits at the headers deadline', async () => { + const fetches = []; + for (let index = 0; index < RENDERED_AVATAR_COUNT; index += 1) { + fetches.push(client.fetch(avatarRequest(index, renderer.signal))); + } + await tick(); + assert.equal(state.started.length, RENDERED_AVATAR_COUNT); + + for (const entry of state.started) { + entry.answer(); + } + const texts = await Promise.all(fetches.map(async (fetch) => await (await fetch).text())); + assert.deepEqual(texts, new Array(RENDERED_AVATAR_COUNT).fill('avatar-bytes')); + }); + + test('a media request starts at once while earlier response bodies sit unread', async () => { + const {outboundHTTP, state} = createUpstream({streamBodies: true}); + const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE}); + const renderer = new AbortController(); + + for (let index = 0; index < PARKED_MEDIA_BODY_COUNT; index += 1) { + const fetch = client.fetch(avatarRequest(index, renderer.signal)); + await tick(); + state.started.at(-1).answer(); + assert.equal((await fetch).status, 200); + } + + const next = client.fetch(avatarRequest(PARKED_MEDIA_BODY_COUNT, renderer.signal)); + await tick(); + assert.equal(state.started.length, PARKED_MEDIA_BODY_COUNT + 1); + state.started.at(-1).answer(); + assert.equal((await next).status, 200); + renderer.abort(); + }); +}); + +describe('media that a slow or silent self-host never answers is cut off', () => { + test('a silent upstream is aborted at the headers deadline', async () => { mock.timers.enable({apis: ['setTimeout']}); try { - const {outboundHTTP, state} = createSilentUpstream(); + const {outboundHTTP, state} = createUpstream({streamBodies: false}); const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE}); const renderer = new AbortController(); - const dropped = []; - for (let index = 0; index < PROXY_MAX_CONCURRENT_REQUESTS; index += 1) { - const fetch = client.fetch(avatarRequest(index, renderer.signal)); - fetch.catch(() => {}); - dropped.push(fetch); - } + const silent = client.fetch(avatarRequest(0, renderer.signal)); + silent.catch(() => {}); await tick(); - const fresh = client.fetch(avatarRequest(1000, renderer.signal)); - await tick(); - assert.equal(state.started.length, PROXY_MAX_CONCURRENT_REQUESTS); mock.timers.tick(RESOURCE_HEADERS_DEADLINE_MS); - await Promise.allSettled(dropped); - await tick(); - assert.equal(state.aborted, PROXY_MAX_CONCURRENT_REQUESTS); - assert.equal(state.started.length, PROXY_MAX_CONCURRENT_REQUESTS + 1); - state.started.at(-1).answer(); - const response = await fresh; - assert.equal(response.status, 200); - assert.equal(await response.text(), 'avatar-bytes'); + await assert.rejects(silent, (error) => error.name === 'LocalAppProxyResponseHeadersDeadlineError'); + assert.equal(state.aborted, 1); } finally { mock.timers.reset(); } }); - - test('API requests keep flowing while every media permit is held by a silent upstream', async () => { - const {outboundHTTP, state} = createSilentUpstream(); - const media = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE}); - const api = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.API}); - const renderer = new AbortController(); - for (let index = 0; index < PROXY_MAX_CONCURRENT_REQUESTS * 2; index += 1) { - media.fetch(avatarRequest(index, renderer.signal)).catch(() => {}); - } - await tick(); - const apiCall = api.fetch(apiRequest(renderer.signal)); - await tick(); - const started = state.started.at(-1); - assert.equal(started.url, 'https://api.fluxer.app/v1/users/@me'); - assert.equal(started.serviceName, LocalAppProxyTraffic.API); - started.answer(); - assert.equal((await apiCall).status, 200); - renderer.abort(); - }); - - test('the newest queued media request is admitted first so what is on screen loads before what scrolled away', async () => { - const {outboundHTTP, state} = createSilentUpstream(); - const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE}); - const renderer = new AbortController(); - const fetches = []; - for (let index = 0; index < PROXY_MAX_CONCURRENT_REQUESTS + 3; index += 1) { - const fetch = client.fetch(avatarRequest(index, renderer.signal)); - fetch.catch(() => {}); - fetches.push(fetch); - } - await tick(); - state.started[0].answer(); - await (await fetches[0]).text(); - await tick(); - assert.equal( - state.started.at(-1).url, - `https://media.fluxer.app/avatars/${PROXY_MAX_CONCURRENT_REQUESTS + 2}.webp`, - ); - renderer.abort(); - }); }); diff --git a/fluxer_desktop/src/main/SelectedInstanceFetch.ts b/fluxer_desktop/src/main/SelectedInstanceFetch.ts index 932d14773..1cc91ed0e 100644 --- a/fluxer_desktop/src/main/SelectedInstanceFetch.ts +++ b/fluxer_desktop/src/main/SelectedInstanceFetch.ts @@ -15,7 +15,6 @@ const INSTANCE_RESPONSE_MAX_BYTES = 1024 * 1024; const INSTANCE_RESPONSE_MAX_CHUNKS = 1024; const INSTANCE_FETCH_TIMEOUT_MS = 15_000; -const SELECTED_INSTANCE_SERVICE_NAME = 'desktop_selected_instance'; const INSTANCE_RESPONSE_DESCRIPTION = 'Instance response'; const RESET_CONTENT_STATUS = 205; @@ -80,7 +79,6 @@ export class DesktopSelectedInstanceClient { headers: request.headers ?? null, method, originTrust: DesktopOriginTrust.BOUND, - serviceName: SELECTED_INSTANCE_SERVICE_NAME, signal: request.signal ?? null, timeoutMs, url: request.url,