fix: make remote media downloads wait once and stop falling back on real errors

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';
 

packages/core/src/utils/fetch.ts

@@ -197,6 +197,8 @@ function errorResponse<T>(reason: string): FetchResponse<T> {
 	};
 }
 
+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<T>(url: URL, options: RequestInit): Promise<FetchRes
 			});
 
 			// TODO: Make @hs/core fetch timeout configurable
-			request.setTimeout(20_000, () => {
-				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);

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,37 @@ 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<unknown>);
+
+			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('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<unknown>);
+
+			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 () => {

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<unknown>, 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,9 @@ 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
+			// 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',

packages/federation-sdk/src/services/media.service.spec.ts

@@ -0,0 +1,147 @@
+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';
+
+import type { ConfigService } from './config.service';
+import { FederationRequestError } from './federation-request.service';
+import type { FederationRequestService } from './federation-request.service';
+import { MediaService, isUnknownEndpoint } 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<unknown>, 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<string, string> }[] = [];
+		const requestBinaryData = mock(async (_method: string, _server: string, endpoint: string, queryParams?: Record<string, string>) => {
+			calls.push({ endpoint, queryParams });
+			throw unknownEndpoint();
+		});
+
+		await buildService(requestBinaryData)
+			.downloadFromRemoteServer('remote.example', 'abc')
+			.catch(() => undefined);
+
+		expect(calls).toHaveLength(3);
+		for (const call of calls) {
+			// 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);
+		}
+	});
+
+	it('does not let the origin fetch the file from a third server on our behalf', async () => {
+		const calls: { endpoint: string; queryParams?: Record<string, string> }[] = [];
+		const requestBinaryData = mock(async (_method: string, _server: string, endpoint: string, queryParams?: Record<string, string>) => {
+			calls.push({ endpoint, queryParams });
+			throw unknownEndpoint();
+		});
+
+		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('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 }));
+
+		const result = await buildService(requestBinaryData).downloadFromRemoteServer('remote.example', 'abc');
+
+		expect(result).toBe(content);
+		expect(requestBinaryData).toHaveBeenCalledTimes(1);
+	});
+});
+
+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, '<html>Not Found</html>'))).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);
+	});
+});

packages/federation-sdk/src/services/media.service.ts

@@ -1,8 +1,29 @@
-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';
+import { FederationRequestError, FederationRequestService } from './federation-request.service';
+
+const TRANSPORT_HEADROOM_MS = 2_000;
+// 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
+// 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 {
@@ -11,20 +32,34 @@ export class MediaService {
 	constructor(private readonly configService: ConfigService, private readonly federationRequest: FederationRequestService) {}
 
 	async downloadFromRemoteServer(serverName: string, mediaId: string): Promise<Buffer | null> {
-		const endpoints = [
-			`/_matrix/federation/v1/media/download/${mediaId}`,
-			`/_matrix/media/v3/download/${serverName}/${mediaId}`,
-			`/_matrix/media/r0/download/${serverName}/${mediaId}`,
+		const timeoutMs = String(DOWNLOAD_TIMEOUT_MS);
+
+		const endpoints: { path: string; queryParams: Record<string, string> }[] = [
+			{
+				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) {
 				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 });
+				}
 			}
 		}