fix: make S3 resilient for Docker registry — no rate limit, retry on transient Telegram errors

- Remove rate limiting from all S3 endpoints (used by Docker registry
  for concurrent blob pushes — 429 would abort the entire push).
- Add retry with exponential backoff in botPool.forwardToStorage for
  transient Telegram errors (network timeouts, 5xx, socket issues).
- Add retry with exponential backoff in botPool.getFileInfo per bot.
- Introduce isTransientError() pattern matcher covering ~20 transient
  error signatures.
- Fix temp file leak in handlePutObject when storeFileFromTemp throws.
- Fix pre-existing missing botPool namespace on getFileInfo call in
  handleGetMultipartObject.
- Fix route handler return type in PUT handler.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

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<void> => {
-  return new Promise((resolve) => setTimeout(resolve, seconds * 1000));
+const sleep = (ms: number): Promise<void> => {
+  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<ForwardResult> {
-    try {
-      const result = await this.enqueueUpload<TelegramMessageResult>(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<TelegramMessageResult>((activeBot) => {
-          const telegram = activeBot.telegram as unknown as Record<string, SendMethod>;
-          return telegram[sendMethodName](config.storageChatId, filePayload, payload);
+    while (attempt <= MAX_TRANSIENT_RETRIES) {
+      attempt++;
+      try {
+        const result = await this.enqueueUpload<TelegramMessageResult>(async () => {
+          const filePayload = { source: fileChunk, filename: fileName };
+          const sendMethodName = sendMethodMap[fileType] || 'sendDocument';
+          const payload = buildSendPayload(fileType, fileName);
+
+          return this.executeWithBotRetry<TelegramMessageResult>((activeBot) => {
+            const telegram = activeBot.telegram as unknown as Record<string, SendMethod>;
+            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<TelegramFileInfo> {
     let lastError: unknown;
     for (const activeBot of this.bots) {
-      try {
-        const result = await activeBot.telegram.getFile(telegramFileId);
-        const fileData = result as unknown as Omit<TelegramFileInfo, 'bot_token'>;
-        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<TelegramFileInfo, 'bot_token'>;
+          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;
       }
     }
 

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

src/interfaces/http/routes/index.ts

@@ -59,18 +59,18 @@ const _handleMaybeS3Root = (req: Request): Response | Promise<Response> => {
 };
 
 /**
- * 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<Response> => {
-  const handler = () => handleS3Request(req, getS3RouteBucket(req));
-  return withRateLimit(handler as (req: Request) => Promise<Response>)(req);
+const handleS3Direct = (req: Request): Promise<Response> => {
+  return handleS3Request(req, getS3RouteBucket(req));
 };
 
 /**
@@ -108,7 +108,7 @@ export const routes = {
     GET: (req: Request): Promise<Response> => {
       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),