From 508aaa9370a3d5982d2dfb4e92fc1b753b56916b Mon Sep 17 00:00:00 2001 From: Matheus Cardoso Date: Wed, 2 Sep 2026 11:42:02 -0300 Subject: [PATCH 1/5] 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) { From 2317136109edbaaad255ea847404c674515dbee6 Mon Sep 17 00:00:00 2001 From: Matheus Cardoso Date: Wed, 30 Sep 2026 11:25:32 -0300 Subject: [PATCH 2/5] fix: keep media timeout_ms within the federation transport timeout --- packages/core/src/index.ts | 2 +- packages/core/src/utils/fetch.ts | 6 +- .../src/services/media.service.spec.ts | 65 +++++++++++++++---- .../src/services/media.service.ts | 20 ++++-- 4 files changed, 72 insertions(+), 21 deletions(-) diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 2af0df73c..2ef9cb291 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -74,7 +74,7 @@ export { makeUnsignedRequest } from './utils/makeRequest'; export type { FetchResponse, MultipartResult } from './utils/fetch'; -export { fetch } from './utils/fetch'; +export { fetch, FEDERATION_REQUEST_TIMEOUT_MS } from './utils/fetch'; export * from './AsyncDispatcher'; diff --git a/packages/core/src/utils/fetch.ts b/packages/core/src/utils/fetch.ts index 279c43833..d6b26dae6 100644 --- a/packages/core/src/utils/fetch.ts +++ b/packages/core/src/utils/fetch.ts @@ -197,6 +197,8 @@ function errorResponse(reason: string): FetchResponse { }; } +export const FEDERATION_REQUEST_TIMEOUT_MS = 20_000; + // works for both http and https. for https on a multihomed server (same server hosting // multiple homeservers) we must verify the cert with the right SNI (hostname), or else the // cert check fails because we connect through the ip and not the hostname (due to matrix spec). @@ -261,8 +263,8 @@ export async function fetch(url: URL, options: RequestInit): Promise { - request.destroy(new Error('Request timed out after 20s')); + request.setTimeout(FEDERATION_REQUEST_TIMEOUT_MS, () => { + request.destroy(new Error(`Request timed out after ${FEDERATION_REQUEST_TIMEOUT_MS}ms`)); }); request.end(options.body); diff --git a/packages/federation-sdk/src/services/media.service.spec.ts b/packages/federation-sdk/src/services/media.service.spec.ts index 8d27b17f3..71e607a83 100644 --- a/packages/federation-sdk/src/services/media.service.spec.ts +++ b/packages/federation-sdk/src/services/media.service.spec.ts @@ -1,4 +1,6 @@ -import { describe, expect, it, mock } from 'bun:test'; +import { describe, expect, it, mock, spyOn } from 'bun:test'; + +import { FEDERATION_REQUEST_TIMEOUT_MS } from '@rocket.chat/federation-core'; import type { ConfigService } from './config.service'; import type { FederationRequestService } from './federation-request.service'; @@ -21,7 +23,7 @@ describe('MediaService.downloadFromRemoteServer', () => { expect(calls).toHaveLength(3); for (const call of calls) { - expect(call.queryParams?.timeout_ms).toBe('20000'); + expect(call.queryParams?.timeout_ms).toBe('18000'); } }); @@ -56,7 +58,7 @@ describe('MediaService.downloadFromRemoteServer', () => { .downloadFromRemoteServer('remote.example', 'abc') .catch(() => undefined); - expect(calls[0]?.timeout_ms).toBe('20000'); + expect(calls[0]?.timeout_ms).toBe('18000'); } finally { if (previous === undefined) { delete process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS; @@ -66,9 +68,9 @@ describe('MediaService.downloadFromRemoteServer', () => { } }); - it('honours a valid configured timeout', async () => { + it('honours a valid configured timeout that is under the transport limit', async () => { const previous = process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS; - process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS = '45000'; + process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS = '9000'; try { const calls: (Record | undefined)[] = []; @@ -81,7 +83,33 @@ describe('MediaService.downloadFromRemoteServer', () => { .downloadFromRemoteServer('remote.example', 'abc') .catch(() => undefined); - expect(calls[0]?.timeout_ms).toBe('45000'); + expect(calls[0]?.timeout_ms).toBe('9000'); + } finally { + if (previous === undefined) { + delete process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS; + } else { + process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS = previous; + } + } + }); + + const countTimeoutWarnings = async (raw: string) => { + const previous = process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS; + process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS = raw; + + try { + const service = buildService( + mock(async () => { + throw new Error('not found'); + }), + ); + const { logger } = service as unknown as { logger: { warn: (...args: unknown[]) => void } }; + const warn = spyOn(logger, 'warn').mockImplementation(() => undefined); + + await service.downloadFromRemoteServer('remote.example', 'abc').catch(() => undefined); + await service.downloadFromRemoteServer('remote.example', 'abc').catch(() => undefined); + + return warn.mock.calls.length; } finally { if (previous === undefined) { delete process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS; @@ -89,6 +117,14 @@ describe('MediaService.downloadFromRemoteServer', () => { process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS = previous; } } + }; + + it('warns once when the configured timeout is capped', async () => { + expect(await countTimeoutWarnings('60000')).toBe(1); + }); + + it('does not warn when the configured timeout fits', async () => { + expect(await countTimeoutWarnings('9000')).toBe(0); }); it('still returns the content from the first endpoint that answers', async () => { @@ -103,17 +139,20 @@ describe('MediaService.downloadFromRemoteServer', () => { }); 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('defaults to the longest wait the transport can actually honour', () => { + expect(resolveDownloadTimeoutMs(undefined)).toBe(18_000); + expect(resolveDownloadTimeoutMs('')).toBe(18_000); + expect(resolveDownloadTimeoutMs(' ')).toBe(18_000); }); - it('caps the wait so a request cannot be held open indefinitely', () => { - expect(resolveDownloadTimeoutMs('120000')).toBe(60_000); + // Asking for longer than the transport allows would promise the origin a wait we hang up on. + it('never advertises a longer wait than the transport allows', () => { + expect(resolveDownloadTimeoutMs('120000')).toBe(18_000); + expect(resolveDownloadTimeoutMs('60000')).toBe(18_000); + expect(resolveDownloadTimeoutMs(String(FEDERATION_REQUEST_TIMEOUT_MS))).toBeLessThan(FEDERATION_REQUEST_TIMEOUT_MS); }); - it('accepts a valid override', () => { + it('accepts a valid override below the limit', () => { expect(resolveDownloadTimeoutMs('5000')).toBe(5_000); expect(resolveDownloadTimeoutMs('0')).toBe(0); }); diff --git a/packages/federation-sdk/src/services/media.service.ts b/packages/federation-sdk/src/services/media.service.ts index 145a59bfa..010aa1c18 100644 --- a/packages/federation-sdk/src/services/media.service.ts +++ b/packages/federation-sdk/src/services/media.service.ts @@ -1,11 +1,12 @@ -import { createLogger } from '@rocket.chat/federation-core'; +import { FEDERATION_REQUEST_TIMEOUT_MS, createLogger } from '@rocket.chat/federation-core'; 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; +const TRANSPORT_HEADROOM_MS = 2_000; +const MAX_DOWNLOAD_TIMEOUT_MS = FEDERATION_REQUEST_TIMEOUT_MS - TRANSPORT_HEADROOM_MS; +const DEFAULT_DOWNLOAD_TIMEOUT_MS = MAX_DOWNLOAD_TIMEOUT_MS; export function resolveDownloadTimeoutMs(raw: string | undefined): number { if (!raw?.trim()) { @@ -30,13 +31,22 @@ export class MediaService { private get timeoutMs(): number { if (this.downloadTimeoutMs === undefined) { + const raw = process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS; try { - this.downloadTimeoutMs = resolveDownloadTimeoutMs(process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS); + this.downloadTimeoutMs = resolveDownloadTimeoutMs(raw); + if (Number(raw) > this.downloadTimeoutMs) { + this.logger.warn({ + msg: 'FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS exceeds what the federation transport can wait for, capping it', + value: raw, + cappedMs: this.downloadTimeoutMs, + transportTimeoutMs: FEDERATION_REQUEST_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, + value: raw, defaultMs: DEFAULT_DOWNLOAD_TIMEOUT_MS, err, }); From 27a0b081a07e3af75b3cf2f580ec01ea9bccaee3 Mon Sep 17 00:00:00 2001 From: Matheus Cardoso Date: Wed, 30 Sep 2026 13:08:44 -0300 Subject: [PATCH 3/5] fix: only fall back to legacy media endpoints when the endpoint is unknown --- .../services/federation-request.service.ts | 16 ++- .../src/services/media.service.spec.ts | 102 ++++++++++++++++-- .../src/services/media.service.ts | 22 +++- 3 files changed, 131 insertions(+), 9 deletions(-) diff --git a/packages/federation-sdk/src/services/federation-request.service.ts b/packages/federation-sdk/src/services/federation-request.service.ts index c785ff3cd..46e4ef8cc 100644 --- a/packages/federation-sdk/src/services/federation-request.service.ts +++ b/packages/federation-sdk/src/services/federation-request.service.ts @@ -25,14 +25,22 @@ interface SignedRequest { type HttpMethod = 'GET' | 'POST' | 'PUT' | 'DELETE'; export class FederationRequestError extends Error { + readonly errcode?: string; + constructor(readonly response: FetchResponse, errorText: string) { let errorDetail = errorText; + let errcode: string | undefined; try { - errorDetail = JSON.stringify(JSON.parse(errorText || '')); + const parsed: unknown = JSON.parse(errorText || ''); + errorDetail = JSON.stringify(parsed); + if (parsed && typeof parsed === 'object' && 'errcode' in parsed && typeof parsed.errcode === 'string') { + errcode = parsed.errcode; + } } catch { /* use raw text if parsing fails */ } super(`Federation request failed: ${response.status} ${errorDetail}`); + this.errcode = errcode; } } @@ -93,7 +101,11 @@ export class FederationRequestService { }); if (!response.ok) { - const errorText = await response.text(); + // not response.text(): it yields '' for anything but text/*, and Matrix errors are application/json + const errorText = await response.buffer().then( + (body) => body.toString(), + () => '', + ); const error = new FederationRequestError(response, errorText); this.logger.error({ msg: 'Federation request failed', diff --git a/packages/federation-sdk/src/services/media.service.spec.ts b/packages/federation-sdk/src/services/media.service.spec.ts index 71e607a83..b258723d0 100644 --- a/packages/federation-sdk/src/services/media.service.spec.ts +++ b/packages/federation-sdk/src/services/media.service.spec.ts @@ -1,20 +1,26 @@ import { describe, expect, it, mock, spyOn } from 'bun:test'; import { FEDERATION_REQUEST_TIMEOUT_MS } from '@rocket.chat/federation-core'; +import type { FetchResponse } from '@rocket.chat/federation-core'; import type { ConfigService } from './config.service'; +import { FederationRequestError } from './federation-request.service'; import type { FederationRequestService } from './federation-request.service'; -import { MediaService, resolveDownloadTimeoutMs } from './media.service'; +import { MediaService, isUnknownEndpoint, resolveDownloadTimeoutMs } from './media.service'; const buildService = (requestBinaryData: unknown) => new MediaService({} as ConfigService, { requestBinaryData } as unknown as FederationRequestService); +const federationError = (status: number, body = '') => new FederationRequestError({ status } as FetchResponse, body); + +const unknownEndpoint = () => federationError(404, JSON.stringify({ errcode: 'M_UNRECOGNIZED' })); + 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'); + throw unknownEndpoint(); }); await buildService(requestBinaryData) @@ -31,7 +37,7 @@ describe('MediaService.downloadFromRemoteServer', () => { 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'); + throw unknownEndpoint(); }); await buildService(requestBinaryData) @@ -51,7 +57,7 @@ describe('MediaService.downloadFromRemoteServer', () => { const calls: (Record | undefined)[] = []; const requestBinaryData = mock(async (_method: string, _server: string, _endpoint: string, queryParams?: Record) => { calls.push(queryParams); - throw new Error('not found'); + throw unknownEndpoint(); }); await buildService(requestBinaryData) @@ -76,7 +82,7 @@ describe('MediaService.downloadFromRemoteServer', () => { const calls: (Record | undefined)[] = []; const requestBinaryData = mock(async (_method: string, _server: string, _endpoint: string, queryParams?: Record) => { calls.push(queryParams); - throw new Error('not found'); + throw unknownEndpoint(); }); await buildService(requestBinaryData) @@ -100,7 +106,7 @@ describe('MediaService.downloadFromRemoteServer', () => { try { const service = buildService( mock(async () => { - throw new Error('not found'); + throw unknownEndpoint(); }), ); const { logger } = service as unknown as { logger: { warn: (...args: unknown[]) => void } }; @@ -127,6 +133,71 @@ describe('MediaService.downloadFromRemoteServer', () => { expect(await countTimeoutWarnings('9000')).toBe(0); }); + it('stops at the first endpoint when the origin answers that the media is not uploaded yet', async () => { + const requestBinaryData = mock(async () => { + throw federationError(504, JSON.stringify({ errcode: 'M_NOT_YET_UPLOADED', error: 'Media has not been uploaded yet' })); + }); + + await expect(buildService(requestBinaryData).downloadFromRemoteServer('remote.example', 'abc')).rejects.toThrow( + 'Failed to download media abc from remote.example', + ); + expect(requestBinaryData).toHaveBeenCalledTimes(1); + }); + + it('stops at the first endpoint when the origin does not have the media', async () => { + const requestBinaryData = mock(async () => { + throw federationError(404, JSON.stringify({ errcode: 'M_NOT_FOUND', error: 'Not found' })); + }); + + await buildService(requestBinaryData) + .downloadFromRemoteServer('remote.example', 'abc') + .catch(() => undefined); + + expect(requestBinaryData).toHaveBeenCalledTimes(1); + }); + + it('stops at the first endpoint when the origin could not be reached', async () => { + const requestBinaryData = mock(async () => { + throw new Error('Request timed out after 20000ms'); + }); + + await buildService(requestBinaryData) + .downloadFromRemoteServer('remote.example', 'abc') + .catch(() => undefined); + + expect(requestBinaryData).toHaveBeenCalledTimes(1); + }); + + it('keeps the origin answer as the cause of the failure', async () => { + const answer = federationError(504, JSON.stringify({ errcode: 'M_NOT_YET_UPLOADED' })); + const requestBinaryData = mock(async () => { + throw answer; + }); + + const failure = await buildService(requestBinaryData) + .downloadFromRemoteServer('remote.example', 'abc') + .catch((err: unknown) => err); + + expect((failure as Error).cause).toBe(answer); + }); + + it('moves on to the legacy endpoint when the federation one is not implemented', async () => { + const content = Buffer.from('file-bytes'); + const endpoints: string[] = []; + const requestBinaryData = mock(async (_method: string, _server: string, endpoint: string) => { + endpoints.push(endpoint); + if (endpoint.startsWith('/_matrix/federation/')) { + throw unknownEndpoint(); + } + return { content }; + }); + + const result = await buildService(requestBinaryData).downloadFromRemoteServer('remote.example', 'abc'); + + expect(result).toBe(content); + expect(endpoints).toEqual(['/_matrix/federation/v1/media/download/abc', '/_matrix/media/v3/download/remote.example/abc']); + }); + it('still returns the content from the first endpoint that answers', async () => { const content = Buffer.from('file-bytes'); const requestBinaryData = mock(async () => ({ content })); @@ -163,3 +234,22 @@ describe('resolveDownloadTimeoutMs', () => { } }); }); + +describe('isUnknownEndpoint', () => { + it('recognises the ways a server says it does not implement an endpoint', () => { + expect(isUnknownEndpoint(federationError(404, JSON.stringify({ errcode: 'M_UNRECOGNIZED' })))).toBe(true); + expect(isUnknownEndpoint(federationError(405, JSON.stringify({ errcode: 'M_UNRECOGNIZED' })))).toBe(true); + expect(isUnknownEndpoint(federationError(404, ''))).toBe(true); + expect(isUnknownEndpoint(federationError(404, 'Not Found'))).toBe(true); + expect(isUnknownEndpoint(federationError(400, JSON.stringify({ errcode: 'M_UNRECOGNIZED' })))).toBe(true); + }); + + it('treats every other answer as the origin speaking about the media', () => { + expect(isUnknownEndpoint(federationError(404, JSON.stringify({ errcode: 'M_NOT_FOUND' })))).toBe(false); + expect(isUnknownEndpoint(federationError(504, JSON.stringify({ errcode: 'M_NOT_YET_UPLOADED' })))).toBe(false); + expect(isUnknownEndpoint(federationError(502, ''))).toBe(false); + expect(isUnknownEndpoint(federationError(400, JSON.stringify({ errcode: 'M_BAD_JSON' })))).toBe(false); + expect(isUnknownEndpoint(federationError(403, JSON.stringify({ errcode: 'M_FORBIDDEN' })))).toBe(false); + expect(isUnknownEndpoint(new Error('socket hang up'))).toBe(false); + }); +}); diff --git a/packages/federation-sdk/src/services/media.service.ts b/packages/federation-sdk/src/services/media.service.ts index 010aa1c18..9cb9d68c8 100644 --- a/packages/federation-sdk/src/services/media.service.ts +++ b/packages/federation-sdk/src/services/media.service.ts @@ -2,7 +2,7 @@ import { FEDERATION_REQUEST_TIMEOUT_MS, createLogger } from '@rocket.chat/federa import { singleton } from 'tsyringe'; import { ConfigService } from './config.service'; -import { FederationRequestService } from './federation-request.service'; +import { FederationRequestError, FederationRequestService } from './federation-request.service'; const TRANSPORT_HEADROOM_MS = 2_000; const MAX_DOWNLOAD_TIMEOUT_MS = FEDERATION_REQUEST_TIMEOUT_MS - TRANSPORT_HEADROOM_MS; @@ -21,6 +21,23 @@ export function resolveDownloadTimeoutMs(raw: string | undefined): number { return Math.min(value, MAX_DOWNLOAD_TIMEOUT_MS); } +// Matrix v1.6 answers an unknown endpoint with 404/405 M_UNRECOGNIZED; older servers send a +// non-JSON or errcode-less 404, and older Synapse a 400 M_UNRECOGNIZED. Anything else is the +// origin's real answer about this media (e.g. 504 M_NOT_YET_UPLOADED), which the next endpoint +// would only repeat after waiting out timeout_ms again. +export function isUnknownEndpoint(err: unknown): boolean { + if (!(err instanceof FederationRequestError)) { + return false; + } + + const { status } = err.response; + if (status === 404 || status === 405) { + return err.errcode === undefined || err.errcode === 'M_UNRECOGNIZED'; + } + + return status === 400 && err.errcode === 'M_UNRECOGNIZED'; +} + @singleton() export class MediaService { private readonly logger = createLogger('MediaService'); @@ -82,6 +99,9 @@ export class MediaService { return response.content; } catch (err) { this.logger.debug(`Endpoint ${endpoint} failed: ${err instanceof Error ? err.message : String(err)}`); + if (!isUnknownEndpoint(err)) { + throw new Error(`Failed to download media ${mediaId} from ${serverName}`, { cause: err }); + } } } From 09cfa6a712d26a7ac600c3268d6b136cfe37a98a Mon Sep 17 00:00:00 2001 From: Matheus Cardoso Date: Wed, 30 Sep 2026 13:21:58 -0300 Subject: [PATCH 4/5] refactor: drop FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS --- .../federation-request.service.spec.ts | 20 ++- .../src/services/media.service.spec.ts | 118 +----------------- .../src/services/media.service.ts | 48 +------ 3 files changed, 27 insertions(+), 159 deletions(-) diff --git a/packages/federation-sdk/src/services/federation-request.service.spec.ts b/packages/federation-sdk/src/services/federation-request.service.spec.ts index 172a1498a..b4020f7e4 100644 --- a/packages/federation-sdk/src/services/federation-request.service.spec.ts +++ b/packages/federation-sdk/src/services/federation-request.service.spec.ts @@ -4,7 +4,7 @@ import * as core from '@rocket.chat/federation-core'; import * as nacl from 'tweetnacl'; import type { ConfigService } from './config.service'; -import { FederationRequestService } from './federation-request.service'; +import { FederationRequestError, FederationRequestService } from './federation-request.service'; describe('FederationRequestService', async () => { let service: FederationRequestService; @@ -256,6 +256,24 @@ describe('FederationRequestService', async () => { } }); + it('keeps the Matrix error of a JSON error response', async () => { + // like @hs/core fetch: text() only yields text/* bodies, and Matrix errors are application/json + spyOn(core, 'fetch').mockResolvedValue({ + ok: false, + status: 404, + text: async () => '', + buffer: async () => Buffer.from('{"errcode":"M_NOT_FOUND","error":"Not found"}'), + } as unknown as core.FetchResponse); + + const failure = await service + .makeSignedRequest({ method: 'GET', domain: 'target.example.com', uri: '/test/path' }) + .catch((err: unknown) => err); + + expect(failure).toBeInstanceOf(FederationRequestError); + expect((failure as FederationRequestError).errcode).toBe('M_NOT_FOUND'); + expect((failure as FederationRequestError).message).toContain('M_NOT_FOUND'); + }); + it('should handle network errors properly', async () => { globalThis.fetch = Object.assign( async () => { diff --git a/packages/federation-sdk/src/services/media.service.spec.ts b/packages/federation-sdk/src/services/media.service.spec.ts index b258723d0..e0916db33 100644 --- a/packages/federation-sdk/src/services/media.service.spec.ts +++ b/packages/federation-sdk/src/services/media.service.spec.ts @@ -1,4 +1,4 @@ -import { describe, expect, it, mock, spyOn } from 'bun:test'; +import { describe, expect, it, mock } from 'bun:test'; import { FEDERATION_REQUEST_TIMEOUT_MS } from '@rocket.chat/federation-core'; import type { FetchResponse } from '@rocket.chat/federation-core'; @@ -6,7 +6,7 @@ import type { FetchResponse } from '@rocket.chat/federation-core'; import type { ConfigService } from './config.service'; import { FederationRequestError } from './federation-request.service'; import type { FederationRequestService } from './federation-request.service'; -import { MediaService, isUnknownEndpoint, resolveDownloadTimeoutMs } from './media.service'; +import { MediaService, isUnknownEndpoint } from './media.service'; const buildService = (requestBinaryData: unknown) => new MediaService({} as ConfigService, { requestBinaryData } as unknown as FederationRequestService); @@ -29,7 +29,9 @@ describe('MediaService.downloadFromRemoteServer', () => { expect(calls).toHaveLength(3); for (const call of calls) { - expect(call.queryParams?.timeout_ms).toBe('18000'); + // long enough for an upload to finish, short enough that the origin answers before we hang up + expect(Number(call.queryParams?.timeout_ms)).toBeGreaterThan(0); + expect(Number(call.queryParams?.timeout_ms)).toBeLessThan(FEDERATION_REQUEST_TIMEOUT_MS); } }); @@ -49,90 +51,6 @@ describe('MediaService.downloadFromRemoteServer', () => { 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 unknownEndpoint(); - }); - - await buildService(requestBinaryData) - .downloadFromRemoteServer('remote.example', 'abc') - .catch(() => undefined); - - expect(calls[0]?.timeout_ms).toBe('18000'); - } 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 that is under the transport limit', async () => { - const previous = process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS; - process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS = '9000'; - - try { - const calls: (Record | undefined)[] = []; - const requestBinaryData = mock(async (_method: string, _server: string, _endpoint: string, queryParams?: Record) => { - calls.push(queryParams); - throw unknownEndpoint(); - }); - - await buildService(requestBinaryData) - .downloadFromRemoteServer('remote.example', 'abc') - .catch(() => undefined); - - expect(calls[0]?.timeout_ms).toBe('9000'); - } finally { - if (previous === undefined) { - delete process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS; - } else { - process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS = previous; - } - } - }); - - const countTimeoutWarnings = async (raw: string) => { - const previous = process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS; - process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS = raw; - - try { - const service = buildService( - mock(async () => { - throw unknownEndpoint(); - }), - ); - const { logger } = service as unknown as { logger: { warn: (...args: unknown[]) => void } }; - const warn = spyOn(logger, 'warn').mockImplementation(() => undefined); - - await service.downloadFromRemoteServer('remote.example', 'abc').catch(() => undefined); - await service.downloadFromRemoteServer('remote.example', 'abc').catch(() => undefined); - - return warn.mock.calls.length; - } finally { - if (previous === undefined) { - delete process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS; - } else { - process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS = previous; - } - } - }; - - it('warns once when the configured timeout is capped', async () => { - expect(await countTimeoutWarnings('60000')).toBe(1); - }); - - it('does not warn when the configured timeout fits', async () => { - expect(await countTimeoutWarnings('9000')).toBe(0); - }); - it('stops at the first endpoint when the origin answers that the media is not uploaded yet', async () => { const requestBinaryData = mock(async () => { throw federationError(504, JSON.stringify({ errcode: 'M_NOT_YET_UPLOADED', error: 'Media has not been uploaded yet' })); @@ -209,32 +127,6 @@ describe('MediaService.downloadFromRemoteServer', () => { }); }); -describe('resolveDownloadTimeoutMs', () => { - it('defaults to the longest wait the transport can actually honour', () => { - expect(resolveDownloadTimeoutMs(undefined)).toBe(18_000); - expect(resolveDownloadTimeoutMs('')).toBe(18_000); - expect(resolveDownloadTimeoutMs(' ')).toBe(18_000); - }); - - // Asking for longer than the transport allows would promise the origin a wait we hang up on. - it('never advertises a longer wait than the transport allows', () => { - expect(resolveDownloadTimeoutMs('120000')).toBe(18_000); - expect(resolveDownloadTimeoutMs('60000')).toBe(18_000); - expect(resolveDownloadTimeoutMs(String(FEDERATION_REQUEST_TIMEOUT_MS))).toBeLessThan(FEDERATION_REQUEST_TIMEOUT_MS); - }); - - it('accepts a valid override below the limit', () => { - 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'); - } - }); -}); - describe('isUnknownEndpoint', () => { it('recognises the ways a server says it does not implement an endpoint', () => { expect(isUnknownEndpoint(federationError(404, JSON.stringify({ errcode: 'M_UNRECOGNIZED' })))).toBe(true); diff --git a/packages/federation-sdk/src/services/media.service.ts b/packages/federation-sdk/src/services/media.service.ts index 9cb9d68c8..cb010f6dd 100644 --- a/packages/federation-sdk/src/services/media.service.ts +++ b/packages/federation-sdk/src/services/media.service.ts @@ -5,21 +5,8 @@ import { ConfigService } from './config.service'; import { FederationRequestError, FederationRequestService } from './federation-request.service'; const TRANSPORT_HEADROOM_MS = 2_000; -const MAX_DOWNLOAD_TIMEOUT_MS = FEDERATION_REQUEST_TIMEOUT_MS - TRANSPORT_HEADROOM_MS; -const DEFAULT_DOWNLOAD_TIMEOUT_MS = MAX_DOWNLOAD_TIMEOUT_MS; - -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); -} +// the origin has to answer before our transport gives up on the request +const DOWNLOAD_TIMEOUT_MS = FEDERATION_REQUEST_TIMEOUT_MS - TRANSPORT_HEADROOM_MS; // Matrix v1.6 answers an unknown endpoint with 404/405 M_UNRECOGNIZED; older servers send a // non-JSON or errcode-less 404, and older Synapse a 400 M_UNRECOGNIZED. Anything else is the @@ -42,39 +29,10 @@ export function isUnknownEndpoint(err: unknown): boolean { 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) { - const raw = process.env.FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS; - try { - this.downloadTimeoutMs = resolveDownloadTimeoutMs(raw); - if (Number(raw) > this.downloadTimeoutMs) { - this.logger.warn({ - msg: 'FEDERATION_MEDIA_DOWNLOAD_TIMEOUT_MS exceeds what the federation transport can wait for, capping it', - value: raw, - cappedMs: this.downloadTimeoutMs, - transportTimeoutMs: FEDERATION_REQUEST_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: raw, - defaultMs: DEFAULT_DOWNLOAD_TIMEOUT_MS, - err, - }); - } - } - - return this.downloadTimeoutMs; - } - async downloadFromRemoteServer(serverName: string, mediaId: string): Promise { - const timeoutMs = String(this.timeoutMs); + const timeoutMs = String(DOWNLOAD_TIMEOUT_MS); const endpoints: { path: string; queryParams: Record }[] = [ { From b9ce19503c2e7b721992ce81fc00ed81e279d3fa Mon Sep 17 00:00:00 2001 From: Matheus Cardoso Date: Fri, 2 Oct 2026 08:43:09 -0300 Subject: [PATCH 5/5] fix: fail the request when a federation error body cannot be read --- .../src/services/federation-request.service.spec.ts | 13 +++++++++++++ .../src/services/federation-request.service.ts | 6 ++---- 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/packages/federation-sdk/src/services/federation-request.service.spec.ts b/packages/federation-sdk/src/services/federation-request.service.spec.ts index b4020f7e4..3072bcb23 100644 --- a/packages/federation-sdk/src/services/federation-request.service.spec.ts +++ b/packages/federation-sdk/src/services/federation-request.service.spec.ts @@ -274,6 +274,19 @@ describe('FederationRequestService', async () => { expect((failure as FederationRequestError).message).toContain('M_NOT_FOUND'); }); + it('fails with the read error when the error body cannot be read', async () => { + const readError = new Error('Response closed before it finished'); + spyOn(core, 'fetch').mockResolvedValue({ + ok: false, + status: 404, + buffer: async () => { + throw readError; + }, + } as unknown as core.FetchResponse); + + await expect(service.makeSignedRequest({ method: 'GET', domain: 'target.example.com', uri: '/test/path' })).rejects.toBe(readError); + }); + it('should handle network errors properly', async () => { globalThis.fetch = Object.assign( async () => { diff --git a/packages/federation-sdk/src/services/federation-request.service.ts b/packages/federation-sdk/src/services/federation-request.service.ts index 46e4ef8cc..d439c3140 100644 --- a/packages/federation-sdk/src/services/federation-request.service.ts +++ b/packages/federation-sdk/src/services/federation-request.service.ts @@ -102,10 +102,8 @@ export class FederationRequestService { if (!response.ok) { // not response.text(): it yields '' for anything but text/*, and Matrix errors are application/json - const errorText = await response.buffer().then( - (body) => body.toString(), - () => '', - ); + // a body that can't be read must fail the request: read as empty it would lose the errcode + const errorText = (await response.buffer()).toString(); const error = new FederationRequestError(response, errorText); this.logger.error({ msg: 'Federation request failed',