feat: concurrent chunk uploads within single file
Deploy FileDrop / deploy (push) Successful in 41s
Deploy FileDrop / deploy (push) Successful in 41s
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>
This commit is contained in:
@@ -81,6 +81,10 @@ export const uploadFileInTelegramChunks = async (input: {
|
|||||||
|
|
||||||
const stream = createReadStream(input.tempPath, { highWaterMark: chunkSizeBytes });
|
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) {
|
for await (const data of stream) {
|
||||||
const chunk = Buffer.isBuffer(data) ? data : Buffer.from(data as Uint8Array);
|
const chunk = Buffer.isBuffer(data) ? data : Buffer.from(data as Uint8Array);
|
||||||
if (chunk.byteLength === 0) continue;
|
if (chunk.byteLength === 0) continue;
|
||||||
@@ -94,24 +98,41 @@ export const uploadFileInTelegramChunks = async (input: {
|
|||||||
input.compress,
|
input.compress,
|
||||||
input.compressionMinSizeBytes,
|
input.compressionMinSizeBytes,
|
||||||
);
|
);
|
||||||
const forwardResult = await botPool.forwardToStorage(
|
const currentPart = partNumber;
|
||||||
bytes,
|
|
||||||
`${input.partFileNamePrefix}.part-${partNumber}`,
|
|
||||||
'document',
|
|
||||||
);
|
|
||||||
|
|
||||||
parts.push({
|
// Fire upload concurrently — don't await inside the read loop
|
||||||
partNumber,
|
const uploadPromise = botPool
|
||||||
telegramFileId: forwardResult.telegramFileId,
|
.forwardToStorage(bytes, `${input.partFileNamePrefix}.part-${currentPart}`, 'document')
|
||||||
telegramFileUniqueId: forwardResult.telegramFileUniqueId,
|
.then((forwardResult) => {
|
||||||
storageMessageId: forwardResult.storageMessageId,
|
parts.push({
|
||||||
sizeBytes: chunk.byteLength,
|
partNumber: currentPart,
|
||||||
storedSizeBytes: bytes.byteLength,
|
telegramFileId: forwardResult.telegramFileId,
|
||||||
compressionAlgorithm,
|
telegramFileUniqueId: forwardResult.telegramFileUniqueId,
|
||||||
etag: computeHash(chunk),
|
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 {
|
return {
|
||||||
parts,
|
parts,
|
||||||
fileHash: hasher.digest('hex'),
|
fileHash: hasher.digest('hex'),
|
||||||
|
|||||||
Reference in New Issue
Block a user