mirror of
https://github.com/fluxerapp/fluxer
synced 2026-10-17 04:55:42 +09:00
fix(media): keep self-host media loading after restarts (#3283)
This commit is contained in:
1 parent
bd56174870
commit
46a356cc40
59 files changed
+1238
-499
No files matched your search
@@ -508,7 +508,7 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
|
||||
}
|
||||
});
|
||||
|
||||
test('a released binding refuses a re-resolution that changes the address scope', async () => {
|
||||
test('a released cleartext binding that re-resolves to a public address never sends cleartext there', async () => {
|
||||
const {port, server} = await startServer({});
|
||||
await new Promise((resolve) => server.close(resolve));
|
||||
let rebound = false;
|
||||
@@ -528,7 +528,67 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
|
||||
};
|
||||
await assert.rejects(client.fetch(request), /ECONNREFUSED/);
|
||||
rebound = true;
|
||||
await assert.rejects(client.fetch(request), /resolved to both public and non-public addresses/);
|
||||
await assert.rejects(client.fetch(request), /requires https for the publicly routable origin/);
|
||||
outboundHTTP.cleanup();
|
||||
});
|
||||
|
||||
test('a split-horizon origin that moves from its LAN address to a public one is resolved again instead of failing until restart', async () => {
|
||||
const {port, server} = await startServer({});
|
||||
await new Promise((resolve) => server.close(resolve));
|
||||
let rebound = false;
|
||||
let resolutions = 0;
|
||||
const outboundHTTP = new DesktopOutboundHTTP({
|
||||
resolveHostAddresses: async () => {
|
||||
resolutions += 1;
|
||||
return rebound ? ['93.184.216.34'] : ['127.0.0.1'];
|
||||
},
|
||||
});
|
||||
const request = (signal) => ({
|
||||
body: null,
|
||||
expectedOrigin: `https://home.example:${port}`,
|
||||
headers: null,
|
||||
method: 'GET',
|
||||
originTrust: DesktopOriginTrust.BOUND,
|
||||
serviceName: 'test',
|
||||
signal,
|
||||
timeoutMs: 2000,
|
||||
url: `https://home.example:${port}/ping`,
|
||||
});
|
||||
await assert.rejects(outboundHTTP.request(request(null)), /ECONNREFUSED/);
|
||||
rebound = true;
|
||||
const left = new AbortController();
|
||||
left.abort();
|
||||
await assert.rejects(outboundHTTP.request(request(left.signal)), /was aborted by its caller/);
|
||||
assert.equal(resolutions, 2);
|
||||
outboundHTTP.cleanup();
|
||||
});
|
||||
|
||||
test('an address that swallows connections is released when the request gives up before connecting', async () => {
|
||||
let resolutions = 0;
|
||||
const outboundHTTP = new DesktopOutboundHTTP({
|
||||
resolveHostAddresses: async () => {
|
||||
resolutions += 1;
|
||||
return ['10.255.255.1'];
|
||||
},
|
||||
});
|
||||
const request = {
|
||||
body: null,
|
||||
expectedOrigin: 'http://blackhole.example:9',
|
||||
headers: null,
|
||||
method: 'GET',
|
||||
originTrust: DesktopOriginTrust.BOUND,
|
||||
serviceName: 'test',
|
||||
signal: null,
|
||||
timeoutMs: 200,
|
||||
url: 'http://blackhole.example:9/ping',
|
||||
};
|
||||
await assert.rejects(outboundHTTP.request(request));
|
||||
await assert.rejects(outboundHTTP.request(request));
|
||||
assert.equal(
|
||||
resolutions,
|
||||
2,
|
||||
'a binding whose address never answers stays pinned for the process lifetime when a timeout does not release it',
|
||||
);
|
||||
outboundHTTP.cleanup();
|
||||
});
|
||||
|
||||
|
||||
@@ -231,13 +231,6 @@ class DesktopOutboundHTTPOriginNotRegisteredError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
class DesktopOutboundHTTPMixedAddressScopeError extends Error {
|
||||
public constructor(origin: string) {
|
||||
super(`Desktop outbound HTTP origin ${origin} resolved to both public and non-public addresses`);
|
||||
this.name = 'DesktopOutboundHTTPMixedAddressScopeError';
|
||||
}
|
||||
}
|
||||
|
||||
class DesktopOutboundHTTPEmptyResolutionError extends Error {
|
||||
public constructor(origin: string) {
|
||||
super(`Desktop outbound HTTP origin ${origin} resolved to no usable address`);
|
||||
@@ -640,15 +633,6 @@ function requireScope(binding: DesktopOriginAddressBinding, requirement: Desktop
|
||||
}
|
||||
}
|
||||
|
||||
function requireStableScope(
|
||||
previous: DesktopOriginAddressBinding | undefined,
|
||||
next: DesktopOriginAddressBinding,
|
||||
): void {
|
||||
if (previous != null && previous.scope !== next.scope) {
|
||||
throw new DesktopOutboundHTTPMixedAddressScopeError(next.origin);
|
||||
}
|
||||
}
|
||||
|
||||
function isStaleKeepAliveSocketError(error: unknown): boolean {
|
||||
if (error == null || typeof error !== 'object') {
|
||||
return false;
|
||||
@@ -1168,7 +1152,6 @@ export class DesktopOutboundHTTP {
|
||||
const record: DesktopPendingOriginBinding = {
|
||||
operation: this.resolveBinding(origin).then((binding) => {
|
||||
this.requireAdmission();
|
||||
requireStableScope(existing, binding);
|
||||
requireScope(binding, record.requirement);
|
||||
this.bindings.set(origin, binding);
|
||||
return binding;
|
||||
@@ -1271,13 +1254,21 @@ export class DesktopOutboundHTTP {
|
||||
let attemptsRemaining = isReplayableRequest(request) ? DESKTOP_OUTBOUND_HTTP_STALE_SOCKET_ATTEMPTS : 0;
|
||||
let active: http.ClientRequest | null = null;
|
||||
let terminated = false;
|
||||
const releaseUnconnectedBinding = (): void => {
|
||||
const socket = active?.socket;
|
||||
if (socket == null || socket.connecting) {
|
||||
this.markBindingUnreachable(binding);
|
||||
}
|
||||
};
|
||||
const timeout = setTimeout(() => {
|
||||
terminated = true;
|
||||
releaseUnconnectedBinding();
|
||||
active?.destroy(new DesktopOutboundHTTPRequestTimeoutError(target.toString(), request.timeoutMs));
|
||||
}, request.timeoutMs);
|
||||
timeout.unref();
|
||||
const onAbort = (): void => {
|
||||
terminated = true;
|
||||
releaseUnconnectedBinding();
|
||||
active?.destroy(new DesktopOutboundHTTPRequestAbortedError(target.toString()));
|
||||
};
|
||||
request.signal?.addEventListener('abort', onAbort, {once: true});
|
||||
|
||||
@@ -120,6 +120,25 @@ export async function resolveDesktopRuntimePlan({
|
||||
return resolved.plan;
|
||||
}
|
||||
|
||||
export async function restoreRememberedRuntimePlan(instanceKey: string): Promise<LocalAppRuntimePlan | null> {
|
||||
let apiEndpoint: string;
|
||||
try {
|
||||
apiEndpoint = requireDesktopRuntimeAPIEndpoint(instanceKey);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
const remembered = await readLastServedDiscovery(apiEndpoint);
|
||||
if (remembered === null || remembered.plan.instanceKey !== instanceKey) {
|
||||
return null;
|
||||
}
|
||||
await getDesktopOutboundHTTP().registerAnchoredOrigins({
|
||||
anchorOrigin: requireDesktopHTTPOrigin(new URL(apiEndpoint).origin),
|
||||
origins: runtimePlanTrustedHTTPOrigins(remembered.plan),
|
||||
unresolvedAnchorRequirement: remembered.anchorRequirement,
|
||||
});
|
||||
return remembered.plan;
|
||||
}
|
||||
|
||||
export function desktopRuntimeInitialInput(): string | null {
|
||||
return getLaunchInstanceEndpointOverride();
|
||||
}
|
||||
|
||||
@@ -53,7 +53,9 @@ installTestModuleStub(
|
||||
}`,
|
||||
);
|
||||
|
||||
const {resolveDesktopRuntimePlan} = await import('@electron/main/DesktopRuntimeDiscovery');
|
||||
const {resolveDesktopRuntimePlan, restoreRememberedRuntimePlan} = await import(
|
||||
'@electron/main/DesktopRuntimeDiscovery'
|
||||
);
|
||||
const {DEPLOYED_OFFICIAL_DOCUMENT} = await import('@fluxer/instance_bootstrap/src/__tests__/DiscoveryFixtures');
|
||||
const {DESKTOP_RUNTIME_DISCOVERY_UNREACHABLE_ERROR_NAME} = await import(
|
||||
'@fluxer/desktop_ipc/src/LocalAppRuntimeContract'
|
||||
@@ -183,3 +185,33 @@ describe('desktop discovery while the instance is unreachable', () => {
|
||||
assert.equal(markers.size, 0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('a media or API route for a saved instance after a cold start', () => {
|
||||
beforeEach(() => {
|
||||
markers.clear();
|
||||
storage.refuseWrites = false;
|
||||
network.fetched.length = 0;
|
||||
network.resolvable = true;
|
||||
network.registrations.length = 0;
|
||||
network.respond = served(DEPLOYED_OFFICIAL_DOCUMENT);
|
||||
});
|
||||
|
||||
test('restores the plan of an instance this app was served before without fetching discovery', async () => {
|
||||
const online = await resolve();
|
||||
network.fetched.length = 0;
|
||||
network.registrations.length = 0;
|
||||
const restored = await restoreRememberedRuntimePlan(online.instanceKey);
|
||||
assert.equal(restored?.instanceKey, online.instanceKey);
|
||||
assert.deepEqual(restored?.endpoints, online.endpoints);
|
||||
assert.deepEqual(network.fetched, []);
|
||||
assert.equal(network.registrations.length, 1);
|
||||
assert.equal(network.registrations[0].unresolvedAnchorRequirement, 'public');
|
||||
});
|
||||
|
||||
test('never invents a plan for an instance this app was not served', async () => {
|
||||
assert.equal(await restoreRememberedRuntimePlan('https://unknown.example'), null);
|
||||
assert.equal(await restoreRememberedRuntimePlan('not a url'), null);
|
||||
assert.deepEqual(network.fetched, []);
|
||||
assert.deepEqual(network.registrations, []);
|
||||
});
|
||||
});
|
||||
@@ -30,7 +30,7 @@ import {LOCAL_APP_API_PATH_PREFIX} from '@fluxer/desktop_ipc/src/LocalAppRouteCo
|
||||
const log = createChildLogger('LocalAppAPIProxy');
|
||||
|
||||
interface LocalAppAPIProxyRuntimePlans {
|
||||
findPlanForRoute(runtimeKey: string): LocalAppRuntimePlan | null;
|
||||
planForRoute(runtimeKey: string): Promise<LocalAppRuntimePlan | null>;
|
||||
}
|
||||
|
||||
interface DesktopLocalAppAPIProxyRequestHandlerDependencies {
|
||||
@@ -62,7 +62,7 @@ export class DesktopLocalAppAPIProxyRequestHandler {
|
||||
if (route == null) {
|
||||
return localAppProxyNotFoundResponse('Unknown local app API runtime');
|
||||
}
|
||||
const plan = this.runtimePlans.findPlanForRoute(route.runtimeKey);
|
||||
const plan = await this.runtimePlans.planForRoute(route.runtimeKey);
|
||||
if (plan == null) {
|
||||
return localAppProxyNotFoundResponse('Unknown local app API runtime');
|
||||
}
|
||||
|
||||
@@ -3,12 +3,13 @@
|
||||
import {DESKTOP_APP_SCHEME} from '@electron/common/Constants';
|
||||
import {createChildLogger} from '@electron/common/Logger';
|
||||
import {getDesktopOutboundHTTP} from '@electron/main/DesktopOutboundHTTP';
|
||||
import {restoreRememberedRuntimePlan} from '@electron/main/DesktopRuntimeDiscovery';
|
||||
import {recordGatewayOriginAnchor} from '@electron/main/GatewayOriginRegistry';
|
||||
import {readStagedHarvestLocalStorageSync} from '@electron/main/LegacyOriginHarvestStore';
|
||||
import type {LocalAppIndexContext} from '@electron/main/LocalAppFileRequestHandler';
|
||||
import {DesktopLocalAppFiles} from '@electron/main/LocalAppFileResolver';
|
||||
import {getDesktopLocalAppAuthorization} from '@electron/main/LocalAppProtocolAuthorization';
|
||||
import {DesktopLocalAppProxyClient} from '@electron/main/LocalAppProxyClient';
|
||||
import {DesktopLocalAppProxyClient, LocalAppProxyTraffic} from '@electron/main/LocalAppProxyClient';
|
||||
import {DesktopLocalAppRequestHandler} from '@electron/main/LocalAppRequestHandler';
|
||||
import {
|
||||
DesktopLocalAppRuntimePlans,
|
||||
@@ -57,6 +58,7 @@ export class DesktopLocalAppProtocol {
|
||||
private readonly shutdownController = new AbortController();
|
||||
private readonly activeOperations = new Set<Promise<unknown>>();
|
||||
private readonly runtimePlans = new DesktopLocalAppRuntimePlans();
|
||||
private readonly planRestorations = new Map<string, Promise<LocalAppRuntimePlan | null>>();
|
||||
private readonly files = new DesktopLocalAppFiles();
|
||||
private readonly requestHandler: DesktopLocalAppRequestHandler;
|
||||
private readonly stopObservingCommittedModuleFiles: () => void;
|
||||
@@ -69,8 +71,15 @@ export class DesktopLocalAppProtocol {
|
||||
authorization: getDesktopLocalAppAuthorization(),
|
||||
files: this.files,
|
||||
indexContext: () => this.getIndexContext(),
|
||||
proxyClient: new DesktopLocalAppProxyClient({outboundHTTP: getDesktopOutboundHTTP()}),
|
||||
runtimePlans: this.runtimePlans,
|
||||
apiProxyClient: new DesktopLocalAppProxyClient({
|
||||
outboundHTTP: getDesktopOutboundHTTP(),
|
||||
traffic: LocalAppProxyTraffic.API,
|
||||
}),
|
||||
resourceProxyClient: new DesktopLocalAppProxyClient({
|
||||
outboundHTTP: getDesktopOutboundHTTP(),
|
||||
traffic: LocalAppProxyTraffic.REMOTE_RESOURCE,
|
||||
}),
|
||||
runtimePlans: {planForRoute: (runtimeKey) => this.planForRoute(runtimeKey)},
|
||||
shutdownSignal: this.shutdownController.signal,
|
||||
});
|
||||
}
|
||||
@@ -126,6 +135,40 @@ export class DesktopLocalAppProtocol {
|
||||
return this.runtimePlans.findPlanForRoute(runtimeKey);
|
||||
}
|
||||
|
||||
public planForRoute(runtimeKey: string): Promise<LocalAppRuntimePlan | null> {
|
||||
const known = this.runtimePlans.findPlanForRoute(runtimeKey);
|
||||
if (known != null) {
|
||||
return Promise.resolve(known);
|
||||
}
|
||||
let restoration = this.planRestorations.get(runtimeKey);
|
||||
if (restoration == null) {
|
||||
restoration = this.restoreRuntimePlan(runtimeKey).finally(() => {
|
||||
this.planRestorations.delete(runtimeKey);
|
||||
});
|
||||
this.planRestorations.set(runtimeKey, restoration);
|
||||
}
|
||||
return restoration;
|
||||
}
|
||||
|
||||
private async restoreRuntimePlan(runtimeKey: string): Promise<LocalAppRuntimePlan | null> {
|
||||
let plan: LocalAppRuntimePlan | null;
|
||||
try {
|
||||
plan = await restoreRememberedRuntimePlan(runtimeKey);
|
||||
} catch (error) {
|
||||
log.warn('Failed to restore a remembered local app runtime plan', {instanceKey: runtimeKey, error});
|
||||
return null;
|
||||
}
|
||||
if (plan == null || !this.acceptingRequests) {
|
||||
return null;
|
||||
}
|
||||
const known = this.runtimePlans.findPlanForRoute(runtimeKey);
|
||||
if (known != null) {
|
||||
return known;
|
||||
}
|
||||
this.cacheRuntimePlan(plan);
|
||||
return plan;
|
||||
}
|
||||
|
||||
public cacheRuntimePlan(plan: LocalAppRuntimePlan): void {
|
||||
this.requireAdmission();
|
||||
this.requireRuntimeOriginsRegistered(plan);
|
||||
|
||||
@@ -19,7 +19,7 @@ const LOCAL_APP_PROXY_DEADLINE_MS = 5 * 60 * 1000;
|
||||
const LOCAL_APP_PROXY_REQUEST_BODY_MAX_BYTES = 100 * 1024 * 1024;
|
||||
export const LOCAL_APP_PROXY_ACCEPT_ENCODING = 'gzip, br';
|
||||
|
||||
const LOCAL_APP_PROXY_SERVICE_NAME = 'desktop_local_app_proxy';
|
||||
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;
|
||||
@@ -28,6 +28,20 @@ const MINIMUM_RESPONSE_STATUS = 200;
|
||||
const MAXIMUM_RESPONSE_STATUS = 599;
|
||||
const NULL_BODY_STATUSES: ReadonlySet<number> = new Set([204, 205, 304]);
|
||||
|
||||
export const LocalAppProxyTraffic = Object.freeze({
|
||||
API: 'desktop_local_app_proxy',
|
||||
REMOTE_RESOURCE: 'desktop_local_app_resource_proxy',
|
||||
} as const);
|
||||
|
||||
export type LocalAppProxyTraffic = (typeof LocalAppProxyTraffic)[keyof typeof LocalAppProxyTraffic];
|
||||
|
||||
class LocalAppProxyResponseHeadersDeadlineError extends Error {
|
||||
public constructor(timeoutMs: number) {
|
||||
super(`Local app proxy upstream sent no response headers within ${timeoutMs} ms`);
|
||||
this.name = 'LocalAppProxyResponseHeadersDeadlineError';
|
||||
}
|
||||
}
|
||||
|
||||
class LocalAppProxyQueueOverflowError extends Error {
|
||||
public constructor() {
|
||||
super('Local app proxy has too many queued requests');
|
||||
@@ -78,6 +92,11 @@ class LocalAppProxyResponseBodyLease {
|
||||
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<LocalAppProxyConcurrencyPermit> {
|
||||
signal.throwIfAborted();
|
||||
@@ -109,7 +128,7 @@ class LocalAppProxyConcurrencyGate {
|
||||
}
|
||||
|
||||
private release(): void {
|
||||
const next = this.waiting.shift();
|
||||
const next = this.newestFirst ? this.waiting.pop() : this.waiting.shift();
|
||||
if (next != null) {
|
||||
next.admit();
|
||||
return;
|
||||
@@ -120,6 +139,7 @@ class LocalAppProxyConcurrencyGate {
|
||||
|
||||
interface DesktopLocalAppProxyClientDependencies {
|
||||
readonly outboundHTTP: DesktopOutboundHTTP;
|
||||
readonly traffic: LocalAppProxyTraffic;
|
||||
}
|
||||
|
||||
interface LocalAppProxyFetchRequest {
|
||||
@@ -206,23 +226,49 @@ 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;
|
||||
}
|
||||
|
||||
private readonly concurrency = new LocalAppProxyConcurrencyGate();
|
||||
|
||||
public async fetch(request: LocalAppProxyFetchRequest): Promise<Response> {
|
||||
const permit = await this.concurrency.acquire(request.signal);
|
||||
try {
|
||||
return await this.runFetch(request, permit);
|
||||
return await this.runFetchWithinHeadersDeadline(request, permit);
|
||||
} catch (error) {
|
||||
permit.release();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async runFetchWithinHeadersDeadline(
|
||||
request: LocalAppProxyFetchRequest,
|
||||
permit: LocalAppProxyConcurrencyPermit,
|
||||
): Promise<Response> {
|
||||
const deadlineMs = this.responseHeadersDeadlineMs;
|
||||
if (deadlineMs == null || requestHasBody(request.method)) {
|
||||
return await this.runFetch(request, permit);
|
||||
}
|
||||
const deadline = new AbortController();
|
||||
const timer = setTimeout(
|
||||
() => deadline.abort(new LocalAppProxyResponseHeadersDeadlineError(deadlineMs)),
|
||||
deadlineMs,
|
||||
);
|
||||
timer.unref();
|
||||
try {
|
||||
return await this.runFetch({...request, signal: AbortSignal.any([request.signal, deadline.signal])}, permit);
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
private async runFetch(
|
||||
request: LocalAppProxyFetchRequest,
|
||||
permit: LocalAppProxyConcurrencyPermit,
|
||||
@@ -236,7 +282,7 @@ export class DesktopLocalAppProxyClient {
|
||||
maximumRequestBodyBytes: LOCAL_APP_PROXY_REQUEST_BODY_MAX_BYTES,
|
||||
method: request.method,
|
||||
originTrust: DesktopOriginTrust.REGISTERED,
|
||||
serviceName: LOCAL_APP_PROXY_SERVICE_NAME,
|
||||
serviceName: this.traffic,
|
||||
signal: request.signal,
|
||||
timeoutMs: LOCAL_APP_PROXY_DEADLINE_MS,
|
||||
url: target.href,
|
||||
|
||||
@@ -3,12 +3,12 @@
|
||||
import assert from 'node:assert/strict';
|
||||
import {Buffer} from 'node:buffer';
|
||||
import {Readable} from 'node:stream';
|
||||
import {describe, test} from 'node:test';
|
||||
import {describe, mock, test} from 'node:test';
|
||||
import {installElectronStub} from './LocalAppTestSupport.test.mjs';
|
||||
|
||||
installElectronStub();
|
||||
|
||||
const {DesktopLocalAppProxyClient} = await import('./LocalAppProxyClient.ts');
|
||||
const {DesktopLocalAppProxyClient, LocalAppProxyTraffic} = await import('./LocalAppProxyClient.ts');
|
||||
|
||||
const PROXY_MAX_CONCURRENT_REQUESTS = 32;
|
||||
const PROXY_MAX_QUEUED_REQUESTS = 4096;
|
||||
@@ -93,7 +93,7 @@ function readAvatar(client, index, signal) {
|
||||
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});
|
||||
const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE});
|
||||
const controller = new AbortController();
|
||||
|
||||
const fetches = [];
|
||||
@@ -113,7 +113,7 @@ describe('the local app proxy bounds its own concurrency instead of overrunning
|
||||
|
||||
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});
|
||||
const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE});
|
||||
const controller = new AbortController();
|
||||
|
||||
const fetches = [];
|
||||
@@ -130,7 +130,7 @@ describe('the local app proxy bounds its own concurrency instead of overrunning
|
||||
|
||||
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});
|
||||
const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE});
|
||||
const held = new AbortController();
|
||||
|
||||
const holding = [];
|
||||
@@ -161,7 +161,7 @@ describe('the local app proxy bounds its own concurrency instead of overrunning
|
||||
|
||||
test('the queue refuses a request past its bound rather than growing without limit', async () => {
|
||||
const {outboundHTTP, state} = createOutboundHTTP();
|
||||
const client = new DesktopLocalAppProxyClient({outboundHTTP});
|
||||
const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE});
|
||||
const controller = new AbortController();
|
||||
|
||||
const accepted = [];
|
||||
@@ -178,3 +178,114 @@ describe('the local app proxy bounds its own concurrency instead of overrunning
|
||||
assert.equal(settled.length, PROXY_MAX_CONCURRENT_REQUESTS + PROXY_MAX_QUEUED_REQUESTS);
|
||||
});
|
||||
});
|
||||
|
||||
const RESOURCE_HEADERS_DEADLINE_MS = 30_000;
|
||||
|
||||
function createSilentUpstream() {
|
||||
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')]),
|
||||
status: 200,
|
||||
statusText: 'OK',
|
||||
url: new URL(request.url),
|
||||
}),
|
||||
};
|
||||
state.started.push(entry);
|
||||
request.signal.addEventListener(
|
||||
'abort',
|
||||
() => {
|
||||
state.aborted += 1;
|
||||
reject(request.signal.reason);
|
||||
},
|
||||
{once: true},
|
||||
);
|
||||
}),
|
||||
};
|
||||
return {outboundHTTP, state};
|
||||
}
|
||||
|
||||
function apiRequest(signal) {
|
||||
return {...avatarRequest(0, signal), targetURL: 'https://api.fluxer.app/v1/users/@me'};
|
||||
}
|
||||
|
||||
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 () => {
|
||||
mock.timers.enable({apis: ['setTimeout']});
|
||||
try {
|
||||
const {outboundHTTP, state} = createSilentUpstream();
|
||||
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);
|
||||
}
|
||||
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');
|
||||
} 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();
|
||||
});
|
||||
});
|
||||
@@ -24,7 +24,7 @@ import {LOCAL_APP_REMOTE_PROXY_PATH_PREFIX} from '@fluxer/desktop_ipc/src/LocalA
|
||||
const log = createChildLogger('LocalAppRemoteResourceProxy');
|
||||
|
||||
interface LocalAppRemoteResourceRuntimePlans {
|
||||
findPlanForRoute(runtimeKey: string): LocalAppRuntimePlan | null;
|
||||
planForRoute(runtimeKey: string): Promise<LocalAppRuntimePlan | null>;
|
||||
}
|
||||
|
||||
interface DesktopLocalAppRemoteResourceRequestHandlerDependencies {
|
||||
@@ -69,7 +69,7 @@ export class DesktopLocalAppRemoteResourceRequestHandler {
|
||||
headers: LOCAL_APP_PROXY_TEXT_RESPONSE_HEADERS,
|
||||
});
|
||||
}
|
||||
const plan = this.runtimePlans.findPlanForRoute(route.runtimeKey);
|
||||
const plan = await this.runtimePlans.planForRoute(route.runtimeKey);
|
||||
if (plan == null) {
|
||||
return localAppProxyNotFoundResponse('Unknown local app resource runtime');
|
||||
}
|
||||
|
||||
@@ -68,8 +68,9 @@ function createHandler(options = {}) {
|
||||
},
|
||||
files: new DesktopLocalAppFiles({rendererRoot}),
|
||||
indexContext: () => ({prebootTheme: null}),
|
||||
proxyClient: createProxyClient(),
|
||||
runtimePlans: {findPlanForRoute: (key) => (key === INSTANCE_KEY ? PLAN : null)},
|
||||
apiProxyClient: createProxyClient(),
|
||||
resourceProxyClient: createProxyClient(),
|
||||
runtimePlans: {planForRoute: async (key) => (key === INSTANCE_KEY ? PLAN : null)},
|
||||
shutdownSignal: new AbortController().signal,
|
||||
...options,
|
||||
});
|
||||
|
||||
@@ -30,14 +30,15 @@ const BLOCKED_PROXY_DOCUMENT_DESTINATIONS: ReadonlySet<string> = new Set([
|
||||
]);
|
||||
|
||||
interface LocalAppRequestRuntimePlans {
|
||||
findPlanForRoute(runtimeKey: string): LocalAppRuntimePlan | null;
|
||||
planForRoute(runtimeKey: string): Promise<LocalAppRuntimePlan | null>;
|
||||
}
|
||||
|
||||
interface DesktopLocalAppRequestHandlerDependencies {
|
||||
readonly authorization: DesktopLocalAppAuthorization;
|
||||
readonly files: DesktopLocalAppFiles;
|
||||
readonly indexContext: () => LocalAppIndexContext;
|
||||
readonly proxyClient: DesktopLocalAppProxyClient;
|
||||
readonly apiProxyClient: DesktopLocalAppProxyClient;
|
||||
readonly resourceProxyClient: DesktopLocalAppProxyClient;
|
||||
readonly runtimePlans: LocalAppRequestRuntimePlans;
|
||||
readonly shutdownSignal: AbortSignal;
|
||||
}
|
||||
@@ -67,11 +68,11 @@ export class DesktopLocalAppRequestHandler {
|
||||
this.authorization = dependencies.authorization;
|
||||
this.shutdownSignal = dependencies.shutdownSignal;
|
||||
this.apiProxyRequestHandler = new DesktopLocalAppAPIProxyRequestHandler({
|
||||
proxyClient: dependencies.proxyClient,
|
||||
proxyClient: dependencies.apiProxyClient,
|
||||
runtimePlans: dependencies.runtimePlans,
|
||||
});
|
||||
this.remoteResourceRequestHandler = new DesktopLocalAppRemoteResourceRequestHandler({
|
||||
proxyClient: dependencies.proxyClient,
|
||||
proxyClient: dependencies.resourceProxyClient,
|
||||
runtimePlans: dependencies.runtimePlans,
|
||||
});
|
||||
this.fileRequestHandler = new DesktopLocalAppFileRequestHandler({
|
||||
|
||||
Reference in new issue
Block a user