From 43e36b76294560dbd31f66b6cd070cb27f7b688e Mon Sep 17 00:00:00 2001 From: asepharyana Date: Thu, 27 Aug 2026 14:55:24 +0700 Subject: [PATCH] perf: parallelize S3 object part fetch + cache Telegram file info Optimize the S3 GET path for chunked/multipart objects and reduce Telegram API round-trips: - object-stream: fetch object parts concurrently (bounded, in-order fan-in) instead of serializing N sequential Telegram CDN fetches. Response latency is now ~the slowest part fetch, not the sum of all part fetches. - bot-pool.getFileInfo: cache file_id -> file_path in the existing in-memory cache so repeated S3 GET/HEAD of the same object skip the Telegram API call (file-controller had its own cache wrapper; the S3 path did not). - chunked-storage + s3-controller: resolve multipart/chunked part CDN URLs concurrently via Promise.all instead of sequentially. - s3-controller multipart: await writer.end() before re-reading the temp part file to avoid a flush race. Adds object-stream-parallel.test.ts covering in-order fan-in, byte ranges, and single-part responses even when the slowest part resolves out of order. --- src/infrastructure/telegram/bot-pool.ts | 16 ++- .../telegram/chunked-storage.ts | 32 +++-- .../http/controllers/s3-controller.ts | 32 +++-- src/interfaces/s3/object-stream.ts | 45 +++++- test/object-stream-parallel.test.ts | 130 ++++++++++++++++++ 5 files changed, 228 insertions(+), 27 deletions(-) create mode 100644 test/object-stream-parallel.test.ts diff --git a/src/infrastructure/telegram/bot-pool.ts b/src/infrastructure/telegram/bot-pool.ts index c311d8f..82fe46b 100644 --- a/src/infrastructure/telegram/bot-pool.ts +++ b/src/infrastructure/telegram/bot-pool.ts @@ -7,6 +7,7 @@ import type { } from '../../domain/ports/telegram-service'; import { config } from '../../env'; import logger from '../../shared/logger/index'; +import { fileInfoCache } from '../cache/index'; import { buildSendPayload, extractUploadedFile, @@ -259,18 +260,31 @@ export class BotPool implements ITelegramService { } async getFileInfo(telegramFileId: string): Promise { + // Telegram file_id → file_path mapping is stable for the lifetime of the + // file. Cache it to avoid a Telegram API round-trip on every S3 GET / HEAD + // of the same object. TTL 1h; a cached (possibly stale) file_path would + // only surface if Telegram recycles a file_id, which it does not for + // documents we own. + const cacheKey = `file_info_${telegramFileId}`; + const cached = fileInfoCache.get(cacheKey) as TelegramFileInfo | null; + if (cached) { + return cached; + } + let lastError: unknown; for (const bot of this.bots) { for (let retry = 0; retry <= MAX_TRANSIENT_RETRIES; retry++) { try { const result = await bot.instance.telegram.getFile(telegramFileId); const fileData = result as unknown as Omit; - return { + const fileInfo: TelegramFileInfo = { file_size: fileData.file_size || 0, mime_type: fileData.mime_type || 'application/octet-stream', file_path: fileData.file_path || '', bot_token: bot.token, }; + fileInfoCache.set(cacheKey, fileInfo); + return fileInfo; } catch (error: unknown) { lastError = error; const errorStr = error instanceof Error ? error.message : String(error); diff --git a/src/infrastructure/telegram/chunked-storage.ts b/src/infrastructure/telegram/chunked-storage.ts index 4f70ede..54fe165 100644 --- a/src/infrastructure/telegram/chunked-storage.ts +++ b/src/infrastructure/telegram/chunked-storage.ts @@ -229,21 +229,27 @@ export class ChunkedStorage { */ async buildChunkedObjectSources(file: FileEntity): Promise { const parts = await this.filePartRepository.listByFileId(file.id); - const sources: ObjectPartSource[] = []; - for (const part of parts) { - const fileInfo = await this.telegramService.getFileInfo(part.telegramFileId); - sources.push({ - telegramFileId: part.telegramFileId, - telegramUrl: `https://api.telegram.org/file/bot${fileInfo.bot_token}/${fileInfo.file_path}`, - sizeBytes: part.sizeBytes, - storedSizeBytes: part.storedSizeBytes, - compressionAlgorithm: part.compressionAlgorithm, - partNumber: part.partNumber, - }); - } + // Resolve every part's Telegram CDN URL concurrently (each is an independent + // getFile call) so total resolution time is ~1 round-trip, not N. + const partInfos = await Promise.all( + parts.map(async (part) => { + const fileInfo = await this.telegramService.getFileInfo(part.telegramFileId); + return { + part, + url: `https://api.telegram.org/file/bot${fileInfo.bot_token}/${fileInfo.file_path}`, + }; + }), + ); - return sources; + return partInfos.map(({ part, url }) => ({ + telegramFileId: part.telegramFileId, + telegramUrl: url, + sizeBytes: part.sizeBytes, + storedSizeBytes: part.storedSizeBytes, + compressionAlgorithm: part.compressionAlgorithm, + partNumber: part.partNumber, + })); } /** diff --git a/src/interfaces/http/controllers/s3-controller.ts b/src/interfaces/http/controllers/s3-controller.ts index ada8aaa..408a947 100644 --- a/src/interfaces/http/controllers/s3-controller.ts +++ b/src/interfaces/http/controllers/s3-controller.ts @@ -702,15 +702,21 @@ const handleGetMultipartObject = async ( } const sources: ObjectPartSource[] = []; - for (const part of parts) { - const fileInfo = await botPool.getFileInfo(part.telegramFileId); - sources.push({ - telegramFileId: part.telegramFileId, - telegramUrl: `https://api.telegram.org/file/bot${fileInfo.bot_token}/${fileInfo.file_path}`, - sizeBytes: part.sizeBytes, - partNumber: part.partNumber, - }); - } + // Resolve all part CDN URLs concurrently (independent getFile calls) so + // assembly latency is ~1 round-trip instead of N. + sources.push( + ...(await Promise.all( + parts.map(async (part) => { + const fileInfo = await botPool.getFileInfo(part.telegramFileId); + return { + telegramFileId: part.telegramFileId, + telegramUrl: `https://api.telegram.org/file/bot${fileInfo.bot_token}/${fileInfo.file_path}`, + sizeBytes: part.sizeBytes, + partNumber: part.partNumber, + }; + }), + )), + ); // H1: Always proxy — never expose bot token in redirect URL @@ -1479,9 +1485,13 @@ const handleUploadPart = async ( hasher.update(chunk); writer.write(chunk); } - writer.end(); + await writer.end(); } catch (error) { - writer.end(); + try { + writer.end(); + } catch { + // ignore during error path + } await cleanupTempFile(tempPath); throw error; } finally { diff --git a/src/interfaces/s3/object-stream.ts b/src/interfaces/s3/object-stream.ts index 6a4513e..91296cb 100644 --- a/src/interfaces/s3/object-stream.ts +++ b/src/interfaces/s3/object-stream.ts @@ -97,12 +97,53 @@ const fetchPartBody = async (planned: PlannedPart): Promise( + items: T[], + limit: number, + fn: (item: T, index: number) => Promise, +): Promise => { + const results: R[] = new Array(items.length); + let next = 0; + + const worker = async (): Promise => { + while (true) { + const idx = next++; + if (idx >= items.length) return; + results[idx] = await fn(items[idx], idx); + } + }; + + const workers = Array.from({ length: Math.max(1, Math.min(limit, items.length)) }, worker); + await Promise.all(workers); + return results; +}; + const concatPartStreams = (plannedParts: PlannedPart[]): ReadableStream => new ReadableStream({ async start(controller) { try { - for (const planned of plannedParts) { - const stream = await fetchPartBody(planned); + // Fetch every part body concurrently (bounded) so the slowest Telegram + // CDN fetch dictates latency instead of the sum of all fetches. + const partStreams = await mapBounded(plannedParts, PART_FETCH_CONCURRENCY, fetchPartBody); + + // Fan-in in part order — object bytes stay correctly ordered. + for (const stream of partStreams) { const reader = stream.getReader(); while (true) { const { value, done } = await reader.read(); diff --git a/test/object-stream-parallel.test.ts b/test/object-stream-parallel.test.ts new file mode 100644 index 0000000..a03d4e7 --- /dev/null +++ b/test/object-stream-parallel.test.ts @@ -0,0 +1,130 @@ +import { afterEach, beforeEach, describe, expect, it, mock } from 'bun:test'; +import { createGetObjectResponse, type ObjectPartSource } from '../src/interfaces/s3/object-stream'; +import type { RangeParseResult } from '../src/interfaces/s3/range'; + +/** + * Stubs `globalThis.fetch` so every Telegram part URL returns a body whose + * bytes encode the part number. The first (and only the first) fetch is + * delayed heavily, so it resolves LAST — proving the response still streams + * parts back in correct order despite out-of-order completion. + */ +const installFetchMock = () => { + const original = globalThis.fetch; + + const fakeFetch = mock((url: string | URL | Request) => { + const u = url.toString(); + const match = u.match(/part(\d+)\.bin/); + const part = match ? Number.parseInt(match[1], 10) : 0; + const bytes = new Uint8Array(8).fill(0); + // Encode part number in the last byte so we can assert ordering. + bytes[7] = part; + + const delay = part === 1 ? 60 : 0; + return new Promise((resolve) => { + setTimeout(() => { + resolve( + new Response(bytes, { + status: 200, + headers: { 'content-length': String(bytes.byteLength) }, + }), + ); + }, delay); + }); + }) as unknown as typeof fetch; + + globalThis.fetch = fakeFetch; + return { original, fakeFetch }; +}; + +const makePart = (partNumber: number, sizeBytes: number): ObjectPartSource => ({ + telegramFileId: `id-${partNumber}`, + telegramUrl: `https://api.telegram.org/file/botTOKEN/part${partNumber}.bin`, + sizeBytes, + partNumber, +}); + +const fullRange = (): RangeParseResult => ({ + type: 'none', +}); + +const collect = async (res: Response): Promise => { + const buf = await res.arrayBuffer(); + return new Uint8Array(buf); +}; + +describe('object-stream parallel fan-in', () => { + let restore: { original: typeof fetch; fakeFetch: unknown } | null = null; + + beforeEach(() => { + restore = installFetchMock(); + }); + + afterEach(() => { + if (restore) { + globalThis.fetch = restore.original; + restore = null; + } + }); + + it('streams multiple parts in order even when part 1 resolves last', async () => { + const parts = [makePart(1, 8), makePart(2, 8), makePart(3, 8)]; + const res = await createGetObjectResponse({ + reqId: 'req-1', + contentType: 'application/octet-stream', + etag: 'etag', + lastModified: new Date('2026-01-01T00:00:00Z'), + totalSize: 24, + parts, + range: fullRange(), + }); + + expect(res.status).toBe(200); + const body = await collect(res); + expect(body.byteLength).toBe(24); + + // Each 8-byte part ends with its part number; order must be 1,2,3. + expect(body[7]).toBe(1); + expect(body[15]).toBe(2); + expect(body[23]).toBe(3); + }); + + it('respects byte range across parts', async () => { + const parts = [makePart(1, 8), makePart(2, 8), makePart(3, 8)]; + const range: RangeParseResult = { type: 'valid', start: 4, end: 19 }; + const res = await createGetObjectResponse({ + reqId: 'req-2', + contentType: 'application/octet-stream', + etag: 'etag', + lastModified: new Date('2026-01-01T00:00:00Z'), + totalSize: 24, + parts, + range, + }); + + expect(res.status).toBe(206); + const body = await collect(res); + // 16 bytes: tail of part1 (byte 4-7 => part1 marker), full part2, head of part3 (0-3) + expect(body.byteLength).toBe(16); + expect(body[3]).toBe(1); // last byte of part 1 (marker) + expect(body[11]).toBe(2); // last byte of part 2 (marker) + // Part 3 contributes only its first 4 bytes (0-3); its marker lives at + // relative byte 7, which is outside the requested range — so no marker here. + expect(body[15]).toBe(0); + }); + + it('handles a single part without error', async () => { + const parts = [makePart(1, 8)]; + const res = await createGetObjectResponse({ + reqId: 'req-3', + contentType: 'application/octet-stream', + etag: 'etag', + lastModified: new Date('2026-01-01T00:00:00Z'), + totalSize: 8, + parts, + range: fullRange(), + }); + const body = await collect(res); + expect(body.byteLength).toBe(8); + expect(body[7]).toBe(1); + }); +});