From 1a38b32fbc50a365edae0b5f21be68bdde0fa14d Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Wed, 29 Jul 2026 12:28:05 +0700
Subject: [PATCH] feat: concurrent chunk uploads within single file

Previously uploadFileInTelegramChunks awaited each chunk's upload
before reading the next, making all chunks sequential within a file.
Now chunks are uploaded concurrently using a managed Set of in-flight
promises with backpressure limiting (2x uploadConcurrency).

This means a single 1GB Docker layer split into 48MB chunks will
have up to 32 chunks uploading simultaneously, not one at a time.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 src/utils/chunked-storage.ts | 49 +++++++++++++++++++++++++-----------
 1 file changed, 35 insertions(+), 14 deletions(-)

diff --git a/src/utils/chunked-storage.ts b/src/utils/chunked-storage.ts
index b1dd62b..74f5d32 100644
--- a/src/utils/chunked-storage.ts
+++ b/src/utils/chunked-storage.ts
@@ -81,6 +81,10 @@ export const uploadFileInTelegramChunks = async (input: {
 
   const stream = createReadStream(input.tempPath, { highWaterMark: chunkSizeBytes });
 
+  // Concurrent upload set: tracks in-flight uploads and limits how many
+  // chunks are being uploaded at once from this single file.
+  const inFlight = new Set<Promise<void>>();
+
   for await (const data of stream) {
     const chunk = Buffer.isBuffer(data) ? data : Buffer.from(data as Uint8Array);
     if (chunk.byteLength === 0) continue;
@@ -94,24 +98,41 @@ export const uploadFileInTelegramChunks = async (input: {
       input.compress,
       input.compressionMinSizeBytes,
     );
-    const forwardResult = await botPool.forwardToStorage(
-      bytes,
-      `${input.partFileNamePrefix}.part-${partNumber}`,
-      'document',
-    );
+    const currentPart = partNumber;
 
-    parts.push({
-      partNumber,
-      telegramFileId: forwardResult.telegramFileId,
-      telegramFileUniqueId: forwardResult.telegramFileUniqueId,
-      storageMessageId: forwardResult.storageMessageId,
-      sizeBytes: chunk.byteLength,
-      storedSizeBytes: bytes.byteLength,
-      compressionAlgorithm,
-      etag: computeHash(chunk),
+    // Fire upload concurrently — don't await inside the read loop
+    const uploadPromise = botPool
+      .forwardToStorage(bytes, `${input.partFileNamePrefix}.part-${currentPart}`, 'document')
+      .then((forwardResult) => {
+        parts.push({
+          partNumber: currentPart,
+          telegramFileId: forwardResult.telegramFileId,
+          telegramFileUniqueId: forwardResult.telegramFileUniqueId,
+          storageMessageId: forwardResult.storageMessageId,
+          sizeBytes: chunk.byteLength,
+          storedSizeBytes: bytes.byteLength,
+          compressionAlgorithm,
+          etag: computeHash(chunk),
+        });
+      });
+
+    // Clean up from in-flight set when done (regardless of success/failure)
+    const trackPromise = uploadPromise.finally(() => {
+      inFlight.delete(trackPromise);
     });
+
+    inFlight.add(trackPromise);
+
+    // Backpressure: if too many chunks are in-flight, wait for one to
+    // finish before reading more — prevents unbounded memory growth.
+    if (inFlight.size >= config.uploadConcurrency * 2) {
+      await Promise.race(inFlight);
+    }
   }
 
+  // Wait for all remaining uploads to finish
+  await Promise.all(inFlight);
+
   return {
     parts,
     fileHash: hasher.digest('hex'),