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
2026-07-29 13:36:16 +07:00
parent d0de4de2d5
commit 2cf6aa3275
1 changed files with 576 additions and 0 deletions
@@ -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<T>(task: () => Promise<T>): Promise<T>;
+```
+
+- [ ] **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) <noreply@anthropic.com>"
+```
+
+### 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<void> => 
+  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<number>): 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<T>(
+    bot: BotEntry,
+    action: (instance: Telegraf, token: string) => Promise<T>,
+  ): Promise<T> {
+    return Promise.race([
+      action(bot.instance, bot.token),
+      new Promise<never>((_, 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<ForwardResult> {
+    let lastError: unknown;
+    const attemptedIndexes = new Set<number>();
+    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<TelegramMessageResult>(
+                bot,
+                (activeBot) => {
+                  const telegram = activeBot.telegram as unknown as Record<string, SendMethod>;
+                  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<TelegramFileInfo> {
+    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<TelegramFileInfo, 'bot_token'>;
+          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) <noreply@anthropic.com>"
+```
+
+### 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) <noreply@anthropic.com>"
+```
+
+### 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) <noreply@anthropic.com>"
+```
+
+### 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) <noreply@anthropic.com>"
+```