From 82750b683f94e277cec683996e32ef98f0372a67 Mon Sep 17 00:00:00 2001 From: Matheus Cardoso Date: Wed, 2 Sep 2026 11:42:02 -0300 Subject: [PATCH] fix: wait for remote media that is still being uploaded --- .../src/services/media.service.spec.ts | 126 ++++++++++++++++++ .../src/services/media.service.ts | 59 +++++++- 2 files changed, 179 insertions(+), 6 deletions(-) create mode 100644 packages/federation-sdk/src/services/media.service.spec.ts diff --git a/packages/federation-sdk/src/services/media.service.spec.ts b/packages/federation-sdk/src/services/media.service.spec.ts new file mode 100644 index 000000000..8d27b17f3 --- /dev/null +++ b/packages/federation-sdk/src/services/media.service.spec.ts @@ -0,0 +1,126 @@ +import { describe, expect, it, mock } from 'bun:test'; + +import type { ConfigService } from './config.service'; +import type { FederationRequestService } from './federation-request.service'; +import { MediaService, resolveDownloadTimeoutMs } from './media.service'; + +const buildService = (requestBinaryData: unknown) => + new MediaService({} as ConfigService, { requestBinaryData } as unknown as FederationRequestService); + +describe('MediaService.downloadFromRemoteServer', () => { + it('asks the origin to wait for an upload that is still being committed', async () => { + const calls: { endpoint: string; queryParams?: Record }[] = []; + const requestBinaryData = mock(async (_method: string, _server: string, endpoint: string, queryParams?: Record) => { + calls.push({ endpoint, queryParams }); + throw new Error('not found'); + }); + + await buildService(requestBinaryData) + .downloadFromRemoteServer('remote.example', 'abc') + .catch(() => undefined); + + expect(calls).toHaveLength(3); + for (const call of calls) { + expect(call.queryParams?.timeout_ms).toBe('20000'); + } + }); + + it('does not let the origin fetch the file from a third server on our behalf', async () => { + const calls: { endpoint: string; queryParams?: Record }[] = []; + const requestBinaryData = mock(async (_method: string, _server: string, endpoint: string, queryParams?: Record) => { + calls.push({ endpoint, queryParams }); + throw new Error('not found'); + }); + + await buildService(requestBinaryData) + .downloadFromRemoteServer('remote.example', 'abc') + .catch(() => undefined); + + expect(calls[0]?.queryParams?.allow_remote).toBeUndefined(); + expect(calls[1]?.queryParams?.allow_remote).toBe('false'); + expect(calls[2]?.queryParams?.allow_remote).toBe('false'); + }); + + it('falls back to the default when the configured timeout is malformed', async () => { + const previous = process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS; + process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS = '20s'; + + try { + const calls: (Record | undefined)[] = []; + const requestBinaryData = mock(async (_method: string, _server: string, _endpoint: string, queryParams?: Record) => { + calls.push(queryParams); + throw new Error('not found'); + }); + + await buildService(requestBinaryData) + .downloadFromRemoteServer('remote.example', 'abc') + .catch(() => undefined); + + expect(calls[0]?.timeout_ms).toBe('20000'); + } finally { + if (previous === undefined) { + delete process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS; + } else { + process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS = previous; + } + } + }); + + it('honours a valid configured timeout', async () => { + const previous = process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS; + process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS = '45000'; + + try { + const calls: (Record | undefined)[] = []; + const requestBinaryData = mock(async (_method: string, _server: string, _endpoint: string, queryParams?: Record) => { + calls.push(queryParams); + throw new Error('not found'); + }); + + await buildService(requestBinaryData) + .downloadFromRemoteServer('remote.example', 'abc') + .catch(() => undefined); + + expect(calls[0]?.timeout_ms).toBe('45000'); + } finally { + if (previous === undefined) { + delete process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS; + } else { + process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS = previous; + } + } + }); + + it('still returns the content from the first endpoint that answers', async () => { + const content = Buffer.from('file-bytes'); + const requestBinaryData = mock(async () => ({ content })); + + const result = await buildService(requestBinaryData).downloadFromRemoteServer('remote.example', 'abc'); + + expect(result).toBe(content); + expect(requestBinaryData).toHaveBeenCalledTimes(1); + }); +}); + +describe('resolveDownloadTimeoutMs', () => { + it('defaults to 20s, matching what other homeservers ask for', () => { + expect(resolveDownloadTimeoutMs(undefined)).toBe(20_000); + expect(resolveDownloadTimeoutMs('')).toBe(20_000); + expect(resolveDownloadTimeoutMs(' ')).toBe(20_000); + }); + + it('caps the wait so a request cannot be held open indefinitely', () => { + expect(resolveDownloadTimeoutMs('120000')).toBe(60_000); + }); + + it('accepts a valid override', () => { + expect(resolveDownloadTimeoutMs('5000')).toBe(5_000); + expect(resolveDownloadTimeoutMs('0')).toBe(0); + }); + + it('rejects a malformed override rather than sending a nonsense timeout', () => { + for (const raw of ['20s', '1.5', '-1', 'abc', 'Infinity']) { + expect(() => resolveDownloadTimeoutMs(raw)).toThrow('Invalid FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS value'); + } + }); +}); diff --git a/packages/federation-sdk/src/services/media.service.ts b/packages/federation-sdk/src/services/media.service.ts index c2663a31b..145a59bfa 100644 --- a/packages/federation-sdk/src/services/media.service.ts +++ b/packages/federation-sdk/src/services/media.service.ts @@ -4,23 +4,70 @@ import { singleton } from 'tsyringe'; import { ConfigService } from './config.service'; import { FederationRequestService } from './federation-request.service'; +const DEFAULT_DOWNLOAD_TIMEOUT_MS = 20_000; +const MAX_DOWNLOAD_TIMEOUT_MS = 60_000; + +export function resolveDownloadTimeoutMs(raw: string | undefined): number { + if (!raw?.trim()) { + return DEFAULT_DOWNLOAD_TIMEOUT_MS; + } + + const value = Number(raw); + if (!Number.isSafeInteger(value) || value < 0) { + throw new Error('Invalid FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS value'); + } + + return Math.min(value, MAX_DOWNLOAD_TIMEOUT_MS); +} + @singleton() export class MediaService { private readonly logger = createLogger('MediaService'); + private downloadTimeoutMs?: number; + constructor(private readonly configService: ConfigService, private readonly federationRequest: FederationRequestService) {} + private get timeoutMs(): number { + if (this.downloadTimeoutMs === undefined) { + try { + this.downloadTimeoutMs = resolveDownloadTimeoutMs(process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS); + } catch (err) { + this.downloadTimeoutMs = DEFAULT_DOWNLOAD_TIMEOUT_MS; + this.logger.warn({ + msg: 'Ignoring invalid FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS, using the default', + value: process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS, + defaultMs: DEFAULT_DOWNLOAD_TIMEOUT_MS, + err, + }); + } + } + + return this.downloadTimeoutMs; + } + async downloadFromRemoteServer(serverName: string, mediaId: string): Promise { - const endpoints = [ - `/_matrix/federation/v1/media/download/${mediaId}`, - `/_matrix/media/v3/download/${serverName}/${mediaId}`, - `/_matrix/media/r0/download/${serverName}/${mediaId}`, + const timeoutMs = String(this.timeoutMs); + + const endpoints: { path: string; queryParams: Record }[] = [ + { + path: `/_matrix/federation/v1/media/download/${mediaId}`, + queryParams: { timeout_ms: timeoutMs }, + }, + { + path: `/_matrix/media/v3/download/${serverName}/${mediaId}`, + queryParams: { allow_remote: 'false', timeout_ms: timeoutMs }, + }, + { + path: `/_matrix/media/r0/download/${serverName}/${mediaId}`, + queryParams: { allow_remote: 'false', timeout_ms: timeoutMs }, + }, ]; - for await (const endpoint of endpoints) { + for await (const { path: endpoint, queryParams } of endpoints) { try { // TODO: Stream remote file downloads instead of buffering the entire file in memory. - const response = await this.federationRequest.requestBinaryData('GET', serverName, endpoint); + const response = await this.federationRequest.requestBinaryData('GET', serverName, endpoint, queryParams); return response.content; } catch (err) {