diff --git a/src/infrastructure/telegram/bot-pool.ts b/src/infrastructure/telegram/bot-pool.ts index e148d02..544705d 100644 --- a/src/infrastructure/telegram/bot-pool.ts +++ b/src/infrastructure/telegram/bot-pool.ts @@ -16,17 +16,66 @@ import { import { enqueueUpload } from './upload-queue'; /** - * Sleep for a given number of seconds. + * Sleep for a given number of milliseconds. * - * Used as a backoff mechanism when all bots in the pool are rate-limited. + * Used as a backoff mechanism when all bots in the pool are rate-limited + * or when retrying transient Telegram API errors. * - * @param seconds - Number of seconds to sleep. + * @param ms - Number of milliseconds to sleep. * @returns A promise that resolves after the specified delay. */ -const sleep = (seconds: number): Promise => { - return new Promise((resolve) => setTimeout(resolve, seconds * 1000)); +const sleep = (ms: number): Promise => { + return new Promise((resolve) => setTimeout(resolve, ms)); }; +/** + * Determines whether an error from the Telegram API is likely transient + * and worth retrying. + * + * Transient telegrams errors include: network timeouts, 5xx server errors, + * and "Too Many Requests" (429) which is already handled by bot rotation + * but is also transient at the network level. + * + * @param error - The caught error object. + * @returns True if the error is likely transient and worth retrying. + */ +const isTransientError = (error: unknown): boolean => { + const str = error instanceof Error ? error.message : String(error); + const transientPatterns = [ + 'retry after', + 'timeout', + 'Timed out', + 'etimedout', + 'econnrefused', + 'econnreset', + 'ECONNREFUSED', + 'ECONNRESET', + 'ETIMEDOUT', + '5xx', + '502', + '503', + '504', + 'Bad Gateway', + 'Service Unavailable', + 'Gateway Timeout', + 'socket hang up', + 'socket closed', + 'fetch failed', + 'network error', + 'network timeout', + 'API closed', + 'read ECONNRESET', + 'write EPIPE', + ]; + return transientPatterns.some((p) => str.toLowerCase().includes(p.toLowerCase())); +}; + +/** + * Maximum number of retries for transient Telegram API errors + * before giving up and propagating the error to the caller. + */ +const MAX_TRANSIENT_RETRIES = 3; + /** * Manages a pool of Telegram bots with automatic rotation and rate-limit handling. * @@ -125,33 +174,57 @@ export class BotPool implements ITelegramService { fileName: string, fileType: string, ): Promise { - try { - const result = await this.enqueueUpload(async () => { - const filePayload = { source: fileChunk, filename: fileName }; - const sendMethodName = sendMethodMap[fileType] || 'sendDocument'; - const payload = buildSendPayload(fileType, fileName); + let lastError: unknown; + let attempt = 0; - return this.executeWithBotRetry((activeBot) => { - const telegram = activeBot.telegram as unknown as Record; - return telegram[sendMethodName](config.storageChatId, filePayload, payload); + while (attempt <= MAX_TRANSIENT_RETRIES) { + attempt++; + try { + const result = await this.enqueueUpload(async () => { + const filePayload = { source: fileChunk, filename: fileName }; + const sendMethodName = sendMethodMap[fileType] || 'sendDocument'; + const payload = buildSendPayload(fileType, fileName); + + return this.executeWithBotRetry((activeBot) => { + const telegram = activeBot.telegram as unknown as Record; + return telegram[sendMethodName](config.storageChatId, filePayload, payload); + }); }); - }); - const uploadedFile = extractUploadedFile(result, fileType); - logger.info('File forwarded to storage', { fileName, message: result.message_id }); + const uploadedFile = extractUploadedFile(result, fileType); + logger.info('File forwarded to storage', { fileName, message: result.message_id }); - return { - telegramFileId: uploadedFile?.file_id || '', - telegramFileUniqueId: uploadedFile?.file_unique_id || '', - storageMessageId: result.message_id, - }; - } catch (error: unknown) { - logger.error('Failed to forward file to storage', { - fileName, - error: error instanceof Error ? error.message : String(error), - }); - throw error; + return { + telegramFileId: uploadedFile?.file_id || '', + telegramFileUniqueId: uploadedFile?.file_unique_id || '', + storageMessageId: result.message_id, + }; + } catch (error: unknown) { + lastError = error; + const errorStr = error instanceof Error ? error.message : String(error); + + if (attempt <= MAX_TRANSIENT_RETRIES && isTransientError(error)) { + const backoffMs = Math.min(1000 * 2 ** attempt, 10_000); + logger.warn(`Transient error forwarding file, retrying (${attempt}/${MAX_TRANSIENT_RETRIES})`, { + fileName, + error: errorStr, + backoffMs, + }); + await sleep(backoffMs); + continue; + } + + logger.error('Failed to forward file to storage', { + fileName, + error: errorStr, + attempt, + }); + throw error; + } } + + // Should not reach here — last iteration throws above + throw lastError; } /** @@ -167,26 +240,39 @@ export class BotPool implements ITelegramService { async getFileInfo(telegramFileId: string): Promise { let lastError: unknown; for (const activeBot of this.bots) { - try { - const result = await activeBot.telegram.getFile(telegramFileId); - const fileData = result as unknown as Omit; - return { - file_size: fileData.file_size || 0, - mime_type: fileData.mime_type || 'application/octet-stream', - file_path: fileData.file_path || '', - bot_token: activeBot.telegram.token, - }; - } catch (error: unknown) { - lastError = error; - const errorStr = error instanceof Error ? error.message : String(error); - if ( - errorStr.includes('wrong file_id') || - errorStr.includes('file is temporarily unavailable') || - errorStr.includes('retry after') - ) { - continue; + for (let retry = 0; retry <= MAX_TRANSIENT_RETRIES; retry++) { + try { + const result = await activeBot.telegram.getFile(telegramFileId); + const fileData = result as unknown as Omit; + return { + file_size: fileData.file_size || 0, + mime_type: fileData.mime_type || 'application/octet-stream', + file_path: fileData.file_path || '', + bot_token: activeBot.telegram.token, + }; + } catch (error: unknown) { + lastError = error; + const errorStr = error instanceof Error ? error.message : String(error); + // Belongs to a different bot — skip to next bot immediately + if ( + errorStr.includes('wrong file_id') || + errorStr.includes('file is temporarily unavailable') + ) { + break; // skip to next bot + } + // Transient — retry on the same bot + if (retry < MAX_TRANSIENT_RETRIES && isTransientError(error)) { + const backoffMs = Math.min(1000 * 2 ** (retry + 1), 5_000); + logger.warn( + `Transient error getting file info, retrying bot ${activeBot.telegram.token.slice(0, 8)}... (${retry + 1}/${MAX_TRANSIENT_RETRIES})`, + { telegramFileId, error: errorStr, backoffMs }, + ); + await sleep(backoffMs); + continue; + } + // Non-transient or exhausted retries — try next bot + break; } - throw error; } } diff --git a/src/interfaces/http/controllers/s3-controller.ts b/src/interfaces/http/controllers/s3-controller.ts index 4319699..6c9d897 100644 --- a/src/interfaces/http/controllers/s3-controller.ts +++ b/src/interfaces/http/controllers/s3-controller.ts @@ -43,7 +43,7 @@ import { parseDeleteObjectsBody, s3ErrorResponse, } from '../../../utils/s3/xml'; -import { forwardToStorage, getFileInfo } from '../../../utils/telegram'; +import { botPool } from '../../../infrastructure/telegram/bot-pool'; /** * The default S3 region returned when no region is explicitly configured. @@ -487,7 +487,7 @@ const handleGetObject = async ( } // Regular Telegram object - const fileInfo = await getFileInfo(file.telegramFileId); + const fileInfo = await botPool.getFileInfo(file.telegramFileId); const redirectUrl = `https://api.telegram.org/file/bot${fileInfo.bot_token}/${fileInfo.file_path}`; const totalSize = file.sizeBytes; @@ -592,7 +592,7 @@ const handleGetMultipartObject = async ( const sources: ObjectPartSource[] = []; for (const part of parts) { - const fileInfo = await getFileInfo(part.telegramFileId); + 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}`, @@ -797,7 +797,12 @@ const handlePutObject = async ( return s3Response(null, 200, reqId, { etag: `"${streamed.fileHash}"` }); } - return await storeFileFromTemp(streamed, key, bucketRecord, contentType, reqId); + try { + return await storeFileFromTemp(streamed, key, bucketRecord, contentType, reqId); + } catch (uploadError) { + await cleanupTempFile(streamed.tempPath); + throw uploadError; + } }; /** @@ -848,7 +853,7 @@ const storeFileFromTemp = async ( return s3Response(null, 200, reqId, { etag: `"${file.fileHash}"` }); } - const forwardResult = await forwardToStorage( + const forwardResult = await botPool.forwardToStorage( createReadStream(streamed.tempPath), partFileNamePrefix, 'document', @@ -1310,7 +1315,7 @@ const handleUploadPart = async ( ); } - const forwardResult = await forwardToStorage( + const forwardResult = await botPool.forwardToStorage( createReadStream(tempPath), `mp-${uploadId}-part-${partNumber}`, 'document', diff --git a/src/interfaces/http/routes/index.ts b/src/interfaces/http/routes/index.ts index 5eeb4d6..cc1ab80 100644 --- a/src/interfaces/http/routes/index.ts +++ b/src/interfaces/http/routes/index.ts @@ -59,18 +59,18 @@ const _handleMaybeS3Root = (req: Request): Response | Promise => { }; /** - * Wraps `handleS3Request` with rate limiting. + * Dispatches an S3 request directly, bypassing rate limiting. * - * S3 API calls (used by Docker registry) are rate-limited per client IP to - * prevent resource exhaustion. The default limit (150 req/60s window) allows - * concurrent layer pushes while still providing protection. + * S3 API calls (used by Docker registry for blob pushes) must not be + * rate-limited — large concurrent layer uploads would hit the limit and + * fail. The Docker registry client retries on 5xx, not 4xx, so a 429 + * would abort the entire push. * * @param req - The incoming S3 request. - * @returns The S3 response or a 429 Too Many Requests error. + * @returns The S3 response. */ -const handleS3WithRateLimit = (req: Request): Promise => { - const handler = () => handleS3Request(req, getS3RouteBucket(req)); - return withRateLimit(handler as (req: Request) => Promise)(req); +const handleS3Direct = (req: Request): Promise => { + return handleS3Request(req, getS3RouteBucket(req)); }; /** @@ -108,7 +108,7 @@ export const routes = { GET: (req: Request): Promise => { const headers = Object.fromEntries(req.headers); if (shouldHandleS3(req, headers)) { - return handleS3WithRateLimit(req); + return handleS3Direct(req); } return handleHome(); }, @@ -118,14 +118,14 @@ export const routes = { } const headers = Object.fromEntries(req.headers); if (shouldHandleS3(req, headers)) { - return handleS3WithRateLimit(req); + return handleS3Direct(req); } - return new Response('Not Allowed', { status: 405 }); + return Promise.resolve(new Response('Not Allowed', { status: 405 })); }, - HEAD: handleS3WithRateLimit, - DELETE: handleS3WithRateLimit, - POST: handleS3WithRateLimit, - OPTIONS: handleS3WithRateLimit, + HEAD: handleS3Direct, + DELETE: handleS3Direct, + POST: handleS3Direct, + OPTIONS: handleS3Direct, }, '/api/v1/auth/login': { POST: withRateLimit(handleLogin),