From adbf9b5efab050b2495c1d8a7800c3f3a2e3154a Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 29 Jul 2026 13:28:38 +0700 Subject: [PATCH] refactor: remove global upload queue Per-bot queues now handle concurrency internally. Co-Authored-By: Claude Opus 5 (1M context) --- src/index.ts | 14 --- src/infrastructure/telegram/upload-queue.ts | 100 -------------------- test/telegramQueue.test.ts | 36 ------- 3 files changed, 150 deletions(-) delete mode 100644 src/infrastructure/telegram/upload-queue.ts delete mode 100644 test/telegramQueue.test.ts diff --git a/src/index.ts b/src/index.ts index c76a989..b072d14 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,7 +1,6 @@ import { serve } from 'bun'; import { config } from './config/index'; import { fileInfoCache } from './infrastructure/cache/index'; -import { clearQueue, getQueueStats, waitForQueue } from './infrastructure/telegram/upload-queue'; import { startBot } from './interfaces/bot/handler'; import { handleS3Request } from './interfaces/http/controllers/s3-controller'; import { cleanupRateLimitCache } from './interfaces/http/middleware/rate-limit'; @@ -67,19 +66,6 @@ const gracefulShutdown = async (signal: string): Promise => { logger.info('Closing HTTP server — no new requests accepted'); server.stop(); - // Drain pending upload queue with a timeout - const { pending, size } = getQueueStats(); - if (pending > 0 || size > 0) { - logger.info('Draining upload queue', { pending, size }); - const drainTimeout = setTimeout(() => { - logger.warn('Upload queue drain timeout — clearing remaining tasks'); - clearQueue(); - }, 30_000); - await waitForQueue(); - clearTimeout(drainTimeout); - logger.info('Upload queue drained'); - } - logger.info('Stopping Telegram bot'); bot.stop(signal); diff --git a/src/infrastructure/telegram/upload-queue.ts b/src/infrastructure/telegram/upload-queue.ts deleted file mode 100644 index 6642538..0000000 --- a/src/infrastructure/telegram/upload-queue.ts +++ /dev/null @@ -1,100 +0,0 @@ -import PQueue from 'p-queue'; -import { config } from '../../env'; -import logger from '../../shared/logger/index'; - -/** - * Maximum number of pending (queued + in-flight) upload tasks before - * new submissions are rejected. Prevents unbounded memory growth when - * Telegram is slow or unavailable. - */ -const MAX_QUEUE_PENDING = 1000; - -/** - * P-queue instance for serialising Telegram upload tasks. - * - * Concurrency is governed by {@link config.uploadConcurrency}. - * Built-in logging emits warnings when the queue grows beyond 5 pending items. - */ -const uploadQueue = new PQueue({ - concurrency: config.uploadConcurrency, -}); - -/* Monitor queue growth and emit warnings for large backlogs */ -uploadQueue.on('add', () => { - const stats = getQueueStats(); - if (stats.size > 5) { - logger.warn('Upload queue building up', { - pending: stats.pending, - size: stats.size, - max: MAX_QUEUE_PENDING, - }); - } -}); - -uploadQueue.on('next', () => { - const stats = getQueueStats(); - logger.debug('Processing next upload', { pending: stats.pending, size: stats.size }); -}); - -/** - * Enqueue an upload task to be executed by the queue. - * - * Tasks are executed in FIFO order, subject to the concurrency limit. - * - * @param task - An async function representing the upload operation. - * @returns A promise that resolves with the task's result. - */ -export const enqueueUpload = (task: () => Promise): Promise => { - const stats = getQueueStats(); - if (stats.pending + stats.size > MAX_QUEUE_PENDING) { - return Promise.reject( - new Error( - `Upload queue full (${stats.pending + stats.size} pending, max ${MAX_QUEUE_PENDING})`, - ), - ); - } - return uploadQueue.add(task); -}; - -/** - * Get current queue statistics. - * - * @returns An object with `pending` (actively executing) and `size` (waiting) counts. - */ -export const getQueueStats = (): { pending: number; size: number } => ({ - pending: uploadQueue.pending, - size: uploadQueue.size, -}); - -/** - * Get the number of items waiting in the queue (not yet started). - * - * @returns The number of queued items. - */ -export const getQueueSize = (): number => uploadQueue.size; - -/** - * Get the number of items currently being processed. - * - * @returns The number of pending (in-flight) items. - */ -export const getPendingCount = (): number => uploadQueue.pending; - -/** - * Clear all pending items and wait for in-flight ones to finish. - * - * @returns A promise that resolves when the queue is idle after clearing. - */ -export const clearQueue = async (): Promise => { - uploadQueue.clear(); - await uploadQueue.onIdle(); -}; - -/** - * Wait for the queue to become idle (all tasks finished). - * - * @returns A promise that resolves when no tasks are pending or in-flight. - */ -export const waitForQueue = async (): Promise => { - await uploadQueue.onIdle(); -}; diff --git a/test/telegramQueue.test.ts b/test/telegramQueue.test.ts deleted file mode 100644 index 9678a24..0000000 --- a/test/telegramQueue.test.ts +++ /dev/null @@ -1,36 +0,0 @@ -import { describe, expect, it } from 'bun:test'; -import { enqueueUpload } from '../src/infrastructure/telegram/upload-queue'; - -describe('Telegram Queue', () => { - it('should process tasks in parallel without limit', async () => { - let activeTasks = 0; - let maxActiveTasks = 0; - - const createTask = (id: number, delayMs: number) => { - return async () => { - activeTasks++; - if (activeTasks > maxActiveTasks) { - maxActiveTasks = activeTasks; - } - - await new Promise((resolve) => setTimeout(resolve, delayMs)); - - activeTasks--; - return id; - }; - }; - - const promises = [ - enqueueUpload(createTask(1, 50)), - enqueueUpload(createTask(2, 50)), - enqueueUpload(createTask(3, 50)), - enqueueUpload(createTask(4, 50)), - ]; - - const results = await Promise.all(promises); - - expect(results).toEqual([1, 2, 3, 4]); - // Concurrency limit is removed, so active tasks should be able to reach 4 (fully parallel) - expect(maxActiveTasks).toBe(4); - }); -});