From a986ce1e08c9e9c5e49782751ba8e904645dcf2d Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 29 Jul 2026 13:25:03 +0700 Subject: [PATCH] =?UTF-8?q?fix:=20per-bot=20queue=20fixes=20=E2=80=94=20ou?= =?UTF-8?q?ter=20loop=20transient=20retry,=20getFileInfo=20logging,=20test?= =?UTF-8?q?=20file,=20safety=20net=20comment,=20empty-bot=20guard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add MAX_OUTER_RETRIES constant and transientAttempts counter for outer-loop retry - Restore getFileInfo transient retry logging with bot identity and fileId - Create test/bot-pool.test.ts with 4 tests for core BotPool behavior - Add empty-bots guard in selectBot() returning null - Add safety net comment and improved logging for outer 429 catch Co-Authored-By: Claude Opus 5 (1M context) --- src/infrastructure/telegram/bot-pool.ts | 26 ++++- test/bot-pool.test.ts | 131 ++++++++++++++++++++++++ 2 files changed, 154 insertions(+), 3 deletions(-) create mode 100644 test/bot-pool.test.ts diff --git a/src/infrastructure/telegram/bot-pool.ts b/src/infrastructure/telegram/bot-pool.ts index bc6e01b..fba5d7e 100644 --- a/src/infrastructure/telegram/bot-pool.ts +++ b/src/infrastructure/telegram/bot-pool.ts @@ -48,6 +48,7 @@ const isTransientError = (error: unknown): boolean => { }; const MAX_TRANSIENT_RETRIES = 3; +const MAX_OUTER_RETRIES = 10; const TELEGRAM_API_TIMEOUT_MS = 120_000; const PER_BOT_CONCURRENCY = 1; @@ -83,6 +84,8 @@ export class BotPool implements ITelegramService { * or in the skip set. */ private selectBot(skipIndexes?: Set): BotEntry | null { + if (this.bots.length === 0) return null; + let best: BotEntry | null = null; let bestPending = Infinity; @@ -133,9 +136,10 @@ export class BotPool implements ITelegramService { ): Promise { let lastError: unknown; const attemptedIndexes = new Set(); + let transientAttempts = 0; - // Outer retry loop — up to 10 attempts across all bots - for (let attempt = 0; attempt < 10; attempt++) { + // Outer retry loop — up to MAX_OUTER_RETRIES attempts across all bots + for (let attempt = 0; attempt < MAX_OUTER_RETRIES; attempt++) { const bot = this.selectBot(attemptedIndexes); if (!bot) { @@ -221,7 +225,19 @@ export class BotPool implements ITelegramService { const retryAfterMatch = errorStr.match(/retry after (\d+)/i); if (retryAfterMatch) { - // Bot was rate-limited — already marked, try next bot + // 429 catch in outer block: serves as a safety net for errors that + // contain "retry after N" wording but were rethrown from the inner + // queue task's fallback path (e.g., non-429 errors with similar text). + logger.warn('Retry-after pattern caught in outer loop (safety net)', { + fileName, + error: errorStr, + }); + continue; + } + + // Transient error at the queue level — retry on next bot + if (transientAttempts < MAX_TRANSIENT_RETRIES && isTransientError(error)) { + transientAttempts++; continue; } @@ -267,6 +283,10 @@ export class BotPool implements ITelegramService { } 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 ${bot.token.slice(0, 8)}... (${retry + 1}/${MAX_TRANSIENT_RETRIES})`, + { telegramFileId, error: errorStr, backoffMs }, + ); await sleep(backoffMs); continue; } diff --git a/test/bot-pool.test.ts b/test/bot-pool.test.ts new file mode 100644 index 0000000..bfe6211 --- /dev/null +++ b/test/bot-pool.test.ts @@ -0,0 +1,131 @@ +import { afterEach, beforeEach, describe, expect, it, mock } from 'bun:test'; + +process.env.BOT_TOKEN = 'bot1:token'; +process.env.ADDITIONAL_BOT_TOKENS = 'bot2:token,bot3:token'; +process.env.STORAGE_CHANNEL_ID = '-1001234567890'; +process.env.BASE_URL = 'https://example.com'; +process.env.DATABASE_URL = 'sqlite://test.db'; +process.env.PORT = '3000'; + +// Track mock queue instances for per-bot assertions +const queueInstances: Array<{ + concurrency: number; + add: ReturnType; + pending: number; + size: number; +}> = []; + +// Mock PQueue so we can verify concurrency +const mockAdd = mock(function addFn(this: any, fn: () => Promise) { + return Promise.resolve().then(() => fn()); +}); + +mock.module('p-queue', () => { + return { + default: mock(function MockQueue(this: any, opts?: { concurrency?: number }) { + const instance = { + concurrency: opts?.concurrency ?? 1, + add: mockAdd, + pending: 0, + size: 0, + }; + queueInstances.push(instance); + return instance; + }), + }; +}); + +// Mock Telegraf — use a class so `new Telegraf(token)` works correctly +const mockTelegramInstances: Record< + string, + { + token: string; + sendDocument: ReturnType; + sendPhoto: ReturnType; + getFile: ReturnType; + } +> = {}; + +class MockTelegraf { + token: string; + telegram: { + token: string; + sendDocument: ReturnType; + sendPhoto: ReturnType; + getFile: ReturnType; + }; + + constructor(token: string) { + this.token = token; + this.telegram = { + token, + sendDocument: mock(() => + Promise.resolve({ + message_id: 1, + document: { file_id: `file_${token}`, file_unique_id: `uniq_${token}` }, + }), + ), + sendPhoto: mock(() => + Promise.resolve({ + message_id: 1, + photo: [{ file_id: `photo_${token}`, file_unique_id: `photo_uniq_${token}` }], + }), + ), + getFile: mock(() => + Promise.resolve({ file_size: 100, mime_type: 'text/plain', file_path: 'path' }), + ), + }; + mockTelegramInstances[token] = this.telegram; + } +} + +mock.module('telegraf', () => ({ + Telegraf: MockTelegraf, +})); + +describe('BotPool', () => { + let BotPool: typeof import('../src/infrastructure/telegram/bot-pool').BotPool; + let botPool: import('../src/infrastructure/telegram/bot-pool').BotPool; + + beforeEach(async () => { + mockAdd.mockClear(); + queueInstances.length = 0; + for (const token of Object.keys(mockTelegramInstances)) { + const tg = mockTelegramInstances[token]; + if (tg) { + tg.sendDocument?.mockClear(); + tg.getFile?.mockClear(); + } + } + const mod = await import('../src/infrastructure/telegram/bot-pool'); + BotPool = mod.BotPool; + botPool = new BotPool(); + }); + + afterEach(() => { + // No module cache cleanup needed — Bun handles import caching correctly + }); + + it('should have correct bot count', () => { + expect(botPool.size).toBe(3); + }); + + it('should have correct effective concurrency', () => { + // 3 bots * 1 concurrency per bot + expect(botPool.getEffectiveConcurrency()).toBe(3); + }); + + it('should forward files through the queue', async () => { + const result = await botPool.forwardToStorage(Buffer.from('test data'), 'test.txt', 'document'); + expect(result.telegramFileId).toBeDefined(); + expect(result.storageMessageId).toBeGreaterThan(0); + }); + + it('should use per-bot queues with concurrency=1', () => { + // Each bot gets its own PQueue instance with concurrency=1 + expect(queueInstances.length).toBe(3); + for (const qi of queueInstances) { + expect(qi.concurrency).toBe(1); + } + }); +});