Pace candidate funnel to send capacity

src/database/candidate-queue-repository.ts

@@ -44,6 +44,7 @@ export interface CandidateEnqueueResult {
   duplicateAuthor: number;
   previouslyProcessed: number;
   previouslyContacted: number;
+  deferredAtCapacity: number;
 }
 
 const ACTIVE_STATUSES: CandidateStatus[] = ['discovered', 'screening', 'ready', 'sending', 'retryable'];
@@ -57,13 +58,19 @@ export class CandidateQueueRepository {
     sourceType: 'image' | 'chat',
     priority: number,
     now: number = Date.now(),
+    maxNewCandidates: number = Number.POSITIVE_INFINITY,
   ): CandidateEnqueueResult {
+    if ((!Number.isInteger(maxNewCandidates) && maxNewCandidates !== Number.POSITIVE_INFINITY)
+      || maxNewCandidates < 0) {
+      throw new Error('Candidate intake capacity must be a non-negative integer');
+    }
     const result: CandidateEnqueueResult = {
       discovered: 0,
       duplicatePost: 0,
       duplicateAuthor: 0,
       previouslyProcessed: 0,
       previouslyContacted: 0,
+      deferredAtCapacity: 0,
     };
     const existingPost = this.db.prepare('SELECT 1 FROM candidate_queue WHERE post_id = ?');
     const processedPost = this.db.prepare('SELECT 1 FROM processed_posts WHERE id = ? LIMIT 1');
@@ -90,7 +97,11 @@ export class CandidateQueueRepository {
     `);
 
     this.db.transaction(() => {
-      for (const post of posts) {
+      for (const [index, post] of posts.entries()) {
+        if (result.discovered >= maxNewCandidates) {
+          result.deferredAtCapacity += posts.length - index;
+          break;
+        }
         if (existingPost.get(post.id)) {
           result.duplicatePost += 1;
           continue;

src/main-loop.ts

@@ -16,6 +16,30 @@ export function currentHourlySendCount(
   return limiter.getStats().messagesThisHour;
 }
 
+export interface CandidateCapacitySnapshot {
+  messagesRemaining: number;
+}
+
+/** Keeps fresh candidate inventory proportional to near-term outbound work. */
+export function calculateCandidateFunnelTarget(
+  accounts: readonly CandidateCapacitySnapshot[],
+  options: {
+    maxReadyCandidates: number;
+    recentWindowMinutes: number;
+    minDelayBetweenMessagesMs: number;
+  },
+): number {
+  if (accounts.length === 0) return 0;
+  const freshnessWindowMs = options.recentWindowMinutes * 60_000;
+  const pacingCapacity = options.minDelayBetweenMessagesMs > 0
+    ? Math.max(1, Math.floor(freshnessWindowMs / options.minDelayBetweenMessagesMs))
+    : Number.POSITIVE_INFINITY;
+  const outboundCapacity = accounts.reduce((total, account) => (
+    total + Math.min(Math.max(0, account.messagesRemaining), pacingCapacity)
+  ), 0);
+  return Math.min(options.maxReadyCandidates, outboundCapacity);
+}
+
 /**
  * Runs a task again only after its previous run has settled. This deliberately
  * replaces setInterval: a slow browser response check must never overlap the
@@ -428,7 +452,6 @@ async function runConcurrentAccountRuntimes(app: AppContainer): Promise<void> {
   const { config, logger, accountRuntimeManager, browserClient, db, webServer, candidateQueueRepo } = app;
   const workerInstance = `candidate-funnel-${process.pid}-${Date.now()}`;
   const leaseMs = 5 * 60_000;
-  const maxActiveCandidates = 500;
   const recovered = candidateQueueRepo.recoverAllLeases();
   app.recipientDeduplicator.recoverStaleReservations(Date.now() - 10 * 60_000);
   logger.info(
@@ -472,6 +495,17 @@ async function runConcurrentAccountRuntimes(app: AppContainer): Promise<void> {
         ));
     };
 
+    const candidateFunnelTarget = (): number => calculateCandidateFunnelTarget(
+      runningRuntimes()
+        .filter((runtime) => runtime.resources.accountAuth.isFullyAuthenticated())
+        .map((runtime) => runtime.resources.rateLimiter.getTimers()),
+      {
+        maxReadyCandidates: config.reddit.maxReadyCandidates,
+        recentWindowMinutes: config.reddit.recentWindowMinutes,
+        minDelayBetweenMessagesMs: config.rateLimit.minDelayBetweenMessagesMs,
+      },
+    );
+
     const discoveryLoop = (async (): Promise<void> => {
       while (!controller.signal.aborted) {
         try {
@@ -479,10 +513,6 @@ async function runConcurrentAccountRuntimes(app: AppContainer): Promise<void> {
             await sleep(1_000, controller.signal).catch(() => undefined);
             continue;
           }
-          if (candidateQueueRepo.countActive() >= maxActiveCandidates) {
-            await sleep(5_000, controller.signal).catch(() => undefined);
-            continue;
-          }
           const enabled = config.reddit.subreddits.filter((subreddit) => subreddit.enabled !== false);
           if (enabled.length === 0) {
             await sleep(5_000, controller.signal).catch(() => undefined);
@@ -493,15 +523,28 @@ async function runConcurrentAccountRuntimes(app: AppContainer): Promise<void> {
             await sleep(5_000, controller.signal).catch(() => undefined);
             continue;
           }
+          const target = candidateFunnelTarget();
+          const active = candidateQueueRepo.countActive();
+          if (target === 0 || active >= target) {
+            await sleep(5_000, controller.signal).catch(() => undefined);
+            continue;
+          }
           const subreddit = enabled[nextSubredditIndex % enabled.length];
           nextSubredditIndex += 1;
           const posts = await scanner.resources.monitorService.fetchCandidates(subreddit);
-          const intake = candidateQueueRepo.enqueue(posts, subreddit.type, subreddit.weight ?? 1);
+          const intake = candidateQueueRepo.enqueue(
+            posts,
+            subreddit.type,
+            subreddit.weight ?? 1,
+            Date.now(),
+            Math.max(0, target - active),
+          );
           logger.info(
             `Candidate intake r/${subreddit.name}: ${posts.length} fetched; ${intake.discovered} queued; `
             + `${intake.duplicatePost} duplicate posts; ${intake.duplicateAuthor} duplicate authors; `
             + `${intake.previouslyProcessed} processed; ${intake.previouslyContacted} contacted; `
-            + `${candidateQueueRepo.countActive()} active in funnel`,
+            + `${intake.deferredAtCapacity} deferred at capacity; `
+            + `${candidateQueueRepo.countActive()}/${target} active in funnel`,
           );
         } catch (error) {
           if (controller.signal.aborted) return;
@@ -526,7 +569,8 @@ async function runConcurrentAccountRuntimes(app: AppContainer): Promise<void> {
         candidateQueueRepo.expireStale(
           Date.now() - config.reddit.recentWindowMinutes * 60_000,
         );
-        if (candidateQueueRepo.countReady() >= config.reddit.maxReadyCandidates) {
+        const target = candidateFunnelTarget();
+        if (target === 0 || candidateQueueRepo.countReady() >= target) {
           await sleep(2_000, controller.signal).catch(() => undefined);
           continue;
         }
@@ -589,7 +633,8 @@ async function runConcurrentAccountRuntimes(app: AppContainer): Promise<void> {
           await sleep(1_000, controller.signal).catch(() => undefined);
           continue;
         }
-        candidateQueueRepo.promoteDueRetries(config.reddit.maxReadyCandidates);
+        const target = candidateFunnelTarget();
+        if (target > 0) candidateQueueRepo.promoteDueRetries(target);
         candidateQueueRepo.recoverExpiredLeases();
         app.recipientDeduplicator.recoverStaleReservations(Date.now() - 10 * 60_000);
         const recipients = await availableRecipients();