perf(worker): run startup upload-recovery batches with bounded concurrency
continuous-integration/drone/push Build is passing

Telegram soft-throttles sustained sequential getMessages calls from a
user account with growing per-call latency (no FLOOD_WAIT, so our retry
wrapper never sees it). On a large package count this made the
once-per-startup destination-message verification pass take 90+ minutes
and block the scheduler/fetch-listener from starting. Running up to 3
batches concurrently cuts wall-clock time well under real per-account
rate limits.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JMm2E4ecmATJo8HuBx92NP
This commit is contained in:
2026-08-21 19:00:21 +02:00
co-authored by Claude Sonnet 5
parent f6381e3178
commit d786f3f23b
+36 -15
View File
@@ -84,25 +84,36 @@ export async function recoverIncompleteUploads(): Promise<void> {
// Batch size for getMessages. TDLib accepts up to ~100 IDs per call. // Batch size for getMessages. TDLib accepts up to ~100 IDs per call.
// Using 100 means 20k packages → ~200 round-trips instead of 20k. // Using 100 means 20k packages → ~200 round-trips instead of 20k.
const BATCH_SIZE = 100; const BATCH_SIZE = 100;
// Telegram soft-throttles sustained sequential history reads from user
// accounts — no FLOOD_WAIT, just growing per-call latency the longer a
// single in-flight request stream runs. A small number of batches in
// flight at once cuts wall-clock time substantially without approaching
// real per-account rate limits.
const CONCURRENCY = 3;
const tdlibClient = client;
const destChannelInfo = destChannel;
const batches: (typeof packages)[] = [];
for (const [, channelPackages] of byChannel) { for (const [, channelPackages] of byChannel) {
// Group packages by destChannelId (already done) — within each group,
// process in batches via getMessages (plural).
for (let i = 0; i < channelPackages.length; i += BATCH_SIZE) { for (let i = 0; i < channelPackages.length; i += BATCH_SIZE) {
const batch = channelPackages.slice(i, i + BATCH_SIZE); batches.push(channelPackages.slice(i, i + BATCH_SIZE));
const batchResults = await verifyMessagesBatch( }
client, }
destChannel.telegramId,
batch.map((p) => p.destMessageId!)
);
for (let j = 0; j < batch.length; j++) { async function processBatch(batch: typeof packages) {
const pkg = batch[j]; const batchResults = await verifyMessagesBatch(
const result = batchResults[j]; tdlibClient,
destChannelInfo.telegramId,
batch.map((p) => p.destMessageId!)
);
if (result.state === "exists") { for (let j = 0; j < batch.length; j++) {
verifiedCount++; const pkg = batch[j];
} else if (result.state === "deleted") { const result = batchResults[j];
if (result.state === "exists") {
verifiedCount++;
} else if (result.state === "deleted") {
log.warn( log.warn(
{ {
packageId: pkg.id, packageId: pkg.id,
@@ -141,10 +152,20 @@ export async function recoverIncompleteUploads(): Promise<void> {
"Could not verify destination message — will retry on next startup" "Could not verify destination message — will retry on next startup"
); );
} }
}
} }
} }
let nextBatchIndex = 0;
async function poolWorker() {
while (nextBatchIndex < batches.length) {
const batch = batches[nextBatchIndex++];
await processBatch(batch);
}
}
await Promise.all(
Array.from({ length: Math.min(CONCURRENCY, batches.length) }, () => poolWorker())
);
log.info( log.info(
{ {
verifiedCount, verifiedCount,