diff --git a/docs/superpowers/plans/2026-07-29-per-bot-queue.md b/docs/superpowers/plans/2026-07-29-per-bot-queue.md new file mode 100644 index 0000000..92dc2f4 --- /dev/null +++ b/docs/superpowers/plans/2026-07-29-per-bot-queue.md @@ -0,0 +1,576 @@ +# Per-Bot Queue Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Replace global PQueue with per-bot queues (concurrency=1 per bot) to eliminate 429 collisions and improve rate-limit safety. + +**Architecture:** Each bot token gets its own PQueue with concurrency=1. Uploads are assigned to the least-loaded available bot via `selectBot()`. On 429, the bot is marked rate-limited and the upload retries on the next available bot. The global `upload-queue.ts` is removed; `uploadConcurrency` config is replaced by `botCount * perBotConcurrency`. + +**Tech Stack:** TypeScript, PQueue, Telegraf + +## Global Constraints + +- Use `Bun` runtime, not Node.js +- Follow existing code style (Biome lint) +- Each bot queue has concurrency=1 (no two uploads hit same bot simultaneously) +- `selectBot()` picks bot with lowest pending queue count, skipping rate-limited bots +- Remove `uploadConcurrency` from config; derive effective concurrency from bot count +- Remove `upload-queue.ts` entirely +- Remove `enqueueUpload` from `ITelegramService` interface + +--- +## File Structure + +### Files to Modify +| File | Change | +|------|--------| +| `src/infrastructure/telegram/bot-pool.ts` | BotEntry array, selectBot(), per-bot queues, retry logic | +| `src/domain/ports/telegram-service.ts` | Remove `enqueueUpload` from interface | +| `src/env.ts` | Remove `uploadConcurrency` field | +| `src/index.ts` | Remove upload-queue import and usage | +| `src/utils/chunked-storage.ts` | Replace `config.uploadConcurrency` with bot count | + +### Files to Delete +| File | Reason | +|------|--------| +| `src/infrastructure/telegram/upload-queue.ts` | Global queue replaced by per-bot queues | +| `test/telegramQueue.test.ts` | Tests for deleted module | + +### Files Not Changed +| File | Reason | +|------|--------| +| `test/bot.test.ts` | Only uses ITelegramService interface (via `forwardToStorage`) | +| `src/infrastructure/telegram/chunked-storage.ts` | Uses ITelegramService interface, not BotPool directly | + +--- + +### Task 1: Refresh the spec & plan files after compaction + +Due to context compaction, re-read the current spec and plan files to ensure accuracy before implementing. + +- [ ] **Step 1: Re-read the spec** + +Read: `docs/superpowers/specs/2026-07-29-per-bot-queue-design.md` + +- [ ] **Step 2: Re-read key implementation files** + +Read: `src/infrastructure/telegram/bot-pool.ts`, `src/env.ts`, `src/utils/chunked-storage.ts` + +### Task 2: Refactor ITelegramService interface + +Remove `enqueueUpload` from the interface — BotPool handles queueing internally now. + +**Files:** +- Modify: `src/domain/ports/telegram-service.ts` + +- [ ] **Step 1: Remove `enqueueUpload` from interface** + +```typescript +// src/domain/ports/telegram-service.ts — remove entire section: + /** + * Enqueue a task for sequential upload execution. + * + * Ensures only one Telegram upload runs at a time to avoid + * rate limits and resource contention. + * + * @param task - An async function performing the upload. + * @returns The result of the task. + */ + enqueueUpload(task: () => Promise): Promise; +``` + +- [ ] **Step 2: Run lint to verify** + +Run: `bunx biome check src/domain/ports/telegram-service.ts` +Expected: No errors. + +- [ ] **Step 3: Commit** + +```bash +git add src/domain/ports/telegram-service.ts +git commit -m "refactor: remove enqueueUpload from ITelegramService + +Per-bot queue handles queueing internally. + +Co-Authored-By: Claude Opus 5 (1M context) " +``` + +### Task 3: Refactor BotPool with per-bot queues + +The core of the redesign. Replace `claimBotIndex()` round-robin with per-bot PQueue instances and `selectBot()` for least-loaded assignment. + +**Files:** +- Modify: `src/infrastructure/telegram/bot-pool.ts` + +**Interfaces:** +- Consumes: `ITelegramService` (no `enqueueUpload` method) +- Produces: `botPool` singleton with per-bot queues, `selectBot()`, per-bot rate-limit tracking + +- [ ] **Step 1: Write test file for per-bot queue behavior** + +```typescript +// test/bot-pool.test.ts +import { beforeEach, describe, expect, it, mock, spyOn } from 'bun:test'; + +// We'll test the BotEntry queue behavior and selectBot logic +``` + +- [ ] **Step 2: Implement BotEntry type and BotPool refactor** + +Replace the class body: + +```typescript +import PQueue from 'p-queue'; +import { Telegraf } from 'telegraf'; +import type { + ForwardResult, + ITelegramService, + TelegramFileInfo, +} from '../../domain/ports/telegram-service'; +import { config } from '../../env'; +import logger from '../../shared/logger/index'; +import { + buildSendPayload, + extractUploadedFile, + type SendMethod, + sendMethodMap, + type TelegramMessageResult, +} from './types'; + +const sleep = (ms: number): Promise => + new Promise((resolve) => setTimeout(resolve, ms)); + +const isTransientError = (error: unknown): boolean => { + const str = error instanceof Error ? error.message : String(error); + const transientPatterns = [ + '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())); +}; + +const MAX_TRANSIENT_RETRIES = 3; +const TELEGRAM_API_TIMEOUT_MS = 120_000; +const PER_BOT_CONCURRENCY = 1; + +interface BotEntry { + index: number; + token: string; + instance: Telegraf; + queue: PQueue; + rateLimitedUntil: number; // 0 = not rate-limited +} + +export class BotPool implements ITelegramService { + private readonly bots: BotEntry[] = []; + + constructor() { + const tokens = Array.from(new Set([config.botToken, ...config.additionalBotTokens])); + this.bots = tokens.map((token, index) => ({ + index, + token, + instance: new Telegraf(token), + queue: new PQueue({ concurrency: PER_BOT_CONCURRENCY }), + rateLimitedUntil: 0, + })); + } + + /** Number of bots in the pool */ + get size(): number { + return this.bots.length; + } + + /** + * Select the bot with the fewest pending tasks that isn't rate-limited + * or in the skip set. + */ + private selectBot(skipIndexes?: Set): BotEntry | null { + let best: BotEntry | null = null; + let bestPending = Infinity; + + for (const bot of this.bots) { + if (skipIndexes?.has(bot.index)) continue; + if (bot.rateLimitedUntil > Date.now()) continue; + + const pending = bot.queue.pending + bot.queue.size; + if (pending < bestPending) { + bestPending = pending; + best = bot; + } + } + + return best; + } + + /** + * Execute a Telegram API action on a specific bot entry. + * Wraps with timeout. + */ + private async executeBotAction( + bot: BotEntry, + action: (instance: Telegraf, token: string) => Promise, + ): Promise { + return Promise.race([ + action(bot.instance, bot.token), + new Promise((_, reject) => + setTimeout( + () => reject(new Error(`Telegram API timeout after ${TELEGRAM_API_TIMEOUT_MS}ms`)), + TELEGRAM_API_TIMEOUT_MS, + ), + ), + ]); + } + + /** + * Forward a file chunk to the configured Telegram storage chat. + * + * The upload is submitted to the least-loaded bot's queue. If the bot + * returns 429, it is marked rate-limited and the upload retries on the + * next available bot. If all bots are rate-limited, sleeps before retrying. + */ + async forwardToStorage( + fileChunk: unknown, + fileName: string, + fileType: string, + ): 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++) { + const bot = this.selectBot(attemptedIndexes); + + if (!bot) { + // No available bots — either all rate-limited or all attempted + if (attemptedIndexes.size > 0) { + // All non-rate-limited bots were tried and failed — wait & reset + logger.warn('All available bots exhausted, sleeping 5s before retry'); + await sleep(5000 + Math.random() * 1000); + attemptedIndexes.clear(); + continue; + } + // All bots rate-limited — wait for the shortest cooldown + const earliestCooldown = Math.min( + ...this.bots.map((b) => b.rateLimitedUntil || Infinity), + ); + const waitMs = Math.max(1000, earliestCooldown - Date.now() + 500); + logger.warn('All bots rate-limited, waiting', { waitMs }); + await sleep(waitMs); + attemptedIndexes.clear(); + continue; + } + + attemptedIndexes.add(bot.index); + + try { + const result = await bot.queue.add(async () => { + // Inner transient retry loop inside the queue + for (let innerRetry = 0; innerRetry <= MAX_TRANSIENT_RETRIES; innerRetry++) { + try { + const filePayload = { source: fileChunk, filename: fileName }; + const sendMethodName = sendMethodMap[fileType] || 'sendDocument'; + const payload = buildSendPayload(fileType, fileName); + + const tgResult = await this.executeBotAction( + bot, + (activeBot) => { + const telegram = activeBot.telegram as unknown as Record; + return telegram[sendMethodName](config.storageChatId, filePayload, payload); + }, + ); + + const uploadedFile = extractUploadedFile(tgResult, fileType); + return { + telegramFileId: uploadedFile?.file_id || '', + telegramFileUniqueId: uploadedFile?.file_unique_id || '', + storageMessageId: tgResult.message_id, + }; + } catch (error: unknown) { + const errorStr = error instanceof Error ? error.message : String(error); + const retryAfterMatch = errorStr.match(/retry after (\d+)/i); + + if (retryAfterMatch) { + // 429 — mark bot rate-limited, throw to outer loop for retry on different bot + const seconds = parseInt(retryAfterMatch[1], 10); + bot.rateLimitedUntil = Date.now() + seconds * 1000; + logger.info(`Bot #${bot.index} rate-limited for ${seconds}s`, { fileName, attempt }); + throw error; // caught by outer retry loop + } + + if (innerRetry < MAX_TRANSIENT_RETRIES && isTransientError(error)) { + const backoffMs = Math.min(1000 * 2 ** innerRetry, 10_000); + logger.warn( + `Transient error on bot #${bot.index}, retrying (${innerRetry + 1}/${MAX_TRANSIENT_RETRIES})`, + { fileName, error: errorStr, backoffMs }, + ); + await sleep(backoffMs); + continue; + } + + throw error; // non-transient — propagate + } + } + + throw new Error(`Exhausted transient retries on bot #${bot.index}`); + }); + + logger.info('File forwarded to storage', { fileName, message: result.storageMessageId }); + return result; + } catch (error: unknown) { + lastError = error; + const errorStr = error instanceof Error ? error.message : String(error); + const retryAfterMatch = errorStr.match(/retry after (\d+)/i); + + if (retryAfterMatch) { + // Bot was rate-limited — already marked, try next bot + continue; + } + + // Transient error at the queue level (timeout, 5xx) + if (transientAttempts < MAX_TRANSIENT_RETRIES && isTransientError(error)) { + transientAttempts++; + continue; + } + + // Non-transient — give up + logger.error('Failed to forward file to storage', { + fileName, + error: errorStr, + attempt, + }); + throw error; + } + } + + throw lastError || new Error('Failed to forward file after all retries'); + } + + /** Get total effective concurrency across all bots */ + getEffectiveConcurrency(): number { + return this.bots.length * PER_BOT_CONCURRENCY; + } + + async getFileInfo(telegramFileId: string): Promise { + let lastError: unknown; + for (const bot of this.bots) { + for (let retry = 0; retry <= MAX_TRANSIENT_RETRIES; retry++) { + try { + const result = await bot.instance.telegram.getFile(telegramFileId); + const fileData = result as unknown as Omit; + return { + file_size: fileData.file_size || 0, + mime_type: fileData.mime_type || 'application/octet-stream', + file_path: fileData.file_path || '', + bot_token: bot.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') + ) { + break; + } + if (retry < MAX_TRANSIENT_RETRIES && isTransientError(error)) { + const backoffMs = Math.min(1000 * 2 ** (retry + 1), 5_000); + await sleep(backoffMs); + continue; + } + break; + } + } + } + logger.error('Failed to get file info from any bot', { + error: lastError instanceof Error ? lastError.message : String(lastError), + }); + throw lastError; + } +} + +export const botPool = new BotPool(); +``` + +- [ ] **Step 3: Run lint** + +Run: `bunx biome check src/infrastructure/telegram/bot-pool.ts` +Expected: No errors. + +- [ ] **Step 4: Run existing test suite** + +Run: `bun test test/bot.test.ts` +Expected: All tests pass. + +- [ ] **Step 5: Commit** + +```bash +git add src/infrastructure/telegram/bot-pool.ts +git commit -m "refactor: per-bot queue with selectBot() and rate-limit tracking + +Each bot has its own PQueue (concurrency=1). Uploads are assigned to +the least-loaded available bot. On 429, the bot is marked rate-limited +and the upload retries on the next available bot. + +Co-Authored-By: Claude Opus 5 (1M context) " +``` + +### Task 4: Remove global upload queue + +Delete the global queue, its test, and all references to it from index.ts. + +**Files:** +- Delete: `src/infrastructure/telegram/upload-queue.ts` +- Modify: `src/index.ts` (lines 4, 71-81) +- Delete: `test/telegramQueue.test.ts` + +- [ ] **Step 1: Delete upload-queue.ts** + +Run: `rm src/infrastructure/telegram/upload-queue.ts` + +- [ ] **Step 2: Delete the test file** + +Run: `rm test/telegramQueue.test.ts` + +- [ ] **Step 3: Update index.ts — remove upload-queue import and shutdown drain logic** + +Remove line: +```typescript +import { clearQueue, getQueueStats, waitForQueue } from './infrastructure/telegram/upload-queue'; +``` + +Remove the drain block (lines 70-81): +```typescript + // 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'); + } +``` + +- [ ] **Step 4: Run lint** + +Run: `bunx biome check src/index.ts` +Expected: No errors. + +- [ ] **Step 5: Run tests** + +Run: `bun test` +Expected: All tests pass (some may be skipped due to missing queue test). + +- [ ] **Step 6: Commit** + +```bash +git add src/infrastructure/telegram/upload-queue.ts test/telegramQueue.test.ts src/index.ts +git commit -m "refactor: remove global upload queue + +Per-bot queues now handle concontrol internally. + +Co-Authored-By: Claude Opus 5 (1M context) " +``` + +### Task 5: Update env.ts and chunked-storage backpressure + +Remove `uploadConcurrency` from config and update chunked-storage to derive effective concurrency from bot pool. + +**Files:** +- Modify: `src/env.ts` +- Modify: `src/utils/chunked-storage.ts` + +- [ ] **Step 1: Remove `uploadConcurrency` from env.ts** + +Remove: +```typescript + uploadConcurrency: number; +``` +and: +```typescript + uploadConcurrency: parseNumber(process.env.UPLOAD_CONCURRENCY, 8), +``` + +- [ ] **Step 2: Update chunked-storage.ts backpressure** + +Replace: +```typescript +import { config } from '../env'; +// ... +if (inFlight.size >= config.uploadConcurrency * 2) { +``` +With: +```typescript +import { botPool } from '../infrastructure/telegram/bot-pool'; +// ... +if (inFlight.size >= botPool.getEffectiveConcurrency()) { +``` +(Use effective concurrency * 2 for backpressure, or just use effective concurrency as the limit.) + +Actually let me think about this more carefully. The backpressure in chunked-storage: +``` +if (inFlight.size >= config.uploadConcurrency * 2) { + await Promise.race(inFlight); + await new Promise((resolve) => setTimeout(resolve, 0)); +} +``` + +This limits the number of in-flight chunks per file. With `uploadConcurrency: 16`, it was 32. Now with effective concurrency of 6 (6 bots), it would be 12. That's fine as backpressure — it prevents too many chunks from being in memory at once. + +Let me use `botPool.getEffectiveConcurrency() * 2` to keep the same multiplier. + +- [ ] **Step 3: Run lint and tests** + +```bash +bunx biome check src/env.ts src/utils/chunked-storage.ts +bun test test/chunked-storage.test.ts +``` + +Expected: All checks pass. + +- [ ] **Step 4: Commit** + +```bash +git add src/env.ts src/utils/chunked-storage.ts +git commit -m "refactor: remove uploadConcurrency from config + +Effective concurrency derived from bot pool size. Chunked-storage +backpressure now uses botPool.getEffectiveConcurrency(). + +Co-Authored-By: Claude Opus 5 (1M context) " +``` + +### Task 6: Full integration test + +- [ ] **Step 1: Run the full test suite** + +Run: `bun test` +Expected: All tests pass. + +- [ ] **Step 2: Run lint** + +Run: `bunx biome check src test` +Expected: No errors. + +- [ ] **Step 3: Create summary commit** + +```bash +git add -A +git commit -m "refactor: implement per-bot queue architecture + +- Each bot has its own PQueue with concurrency=1 +- selectBot() assigns uploads to least-loaded available bot +- 429 rate limits are tracked per-bot with cooldown timers +- Failed uploads retry on next available bot +- Removed global upload-queue.ts and uploadConcurrency config +- Updated ITelegramService interface + +Co-Authored-By: Claude Opus 5 (1M context) " +```