fix: reduce Convex moderation churn causing excessive bandwidth and action compute

convex/_generated/api.d.ts

@@ -20,6 +20,7 @@ import type * as follows from "../follows.js";
 import type * as generatedImages from "../generatedImages.js";
 import type * as http from "../http.js";
 import type * as lib_batchGenerationState from "../lib/batchGenerationState.js";
+import type * as lib_cloudflareWorkerHttp from "../lib/cloudflareWorkerHttp.js";
 import type * as lib_crypto from "../lib/crypto.js";
 import type * as lib_dirtberryCrop from "../lib/dirtberryCrop.js";
 import type * as lib_groq from "../lib/groq.js";
@@ -69,6 +70,7 @@ declare const fullApi: ApiFromModules<{
   generatedImages: typeof generatedImages;
   http: typeof http;
   "lib/batchGenerationState": typeof lib_batchGenerationState;
+  "lib/cloudflareWorkerHttp": typeof lib_cloudflareWorkerHttp;
   "lib/crypto": typeof lib_crypto;
   "lib/dirtberryCrop": typeof lib_dirtberryCrop;
   "lib/groq": typeof lib_groq;

convex/contentAnalysis.test.ts

@@ -1,39 +1,182 @@
 import { describe, expect, it } from "vitest"
 import {
     getNextAnalysisRunDelayMs,
+    getProviderRecoveryDelayMs,
+    shouldSkipAnalyzeRecentImagesSchedule,
     getVisionFailureDispatchStatus,
     shouldRunRecoveryPromptInference,
+    getEffectiveScheduledBackgroundJob,
+    getLatestBackgroundJobLastRunAt,
 } from "./contentAnalysis"
 
 describe("getNextAnalysisRunDelayMs", () => {
     it("schedules the next analysis run when more work is queued", () => {
         expect(
             getNextAnalysisRunDelayMs({
                 queuedImageCount: 2,
-                allProvidersRateLimited: false,
+                providerRecoveryDelayMs: null,
             })
         ).toBe(2100)
     })
 
-    it("schedules another run when providers are rate-limited", () => {
+    it("schedules another run when providers have a recovery delay", () => {
         expect(
             getNextAnalysisRunDelayMs({
                 queuedImageCount: 2,
-                allProvidersRateLimited: true,
+                providerRecoveryDelayMs: 60_000,
             })
-        ).toBe(2100)
+        ).toBe(60_000)
     })
 
     it("does not schedule another run when there is no lookahead work", () => {
         expect(
             getNextAnalysisRunDelayMs({
                 queuedImageCount: 1,
-                allProvidersRateLimited: false,
+                providerRecoveryDelayMs: null,
+            })
+        ).toBeNull()
+    })
+})
+
+describe("getProviderRecoveryDelayMs", () => {
+    it("waits until the earliest provider reset instead of hot-looping", () => {
+        expect(
+            getProviderRecoveryDelayMs({
+                now: 1_000,
+                providerHealths: [
+                    { isAvailable: false, rateLimitedUntil: 11_000 },
+                    { isAvailable: false, rateLimitedUntil: 21_000 },
+                ],
+            })
+        ).toBe(10_000)
+    })
+
+    it("returns the floor delay when reset is too close", () => {
+        expect(
+            getProviderRecoveryDelayMs({
+                now: 1_000,
+                providerHealths: [
+                    { isAvailable: false, rateLimitedUntil: 1_500 },
+                ],
+            })
+        ).toBe(2100)
+    })
+
+    it("returns the floor delay when providers are unavailable without future reset times", () => {
+        expect(
+            getProviderRecoveryDelayMs({
+                now: 1_000,
+                providerHealths: [
+                    { isAvailable: false },
+                    { isAvailable: false, rateLimitedUntil: 500 },
+                ],
+            })
+        ).toBe(2100)
+    })
+
+    it("returns null when a provider is already available", () => {
+        expect(
+            getProviderRecoveryDelayMs({
+                now: 1_000,
+                providerHealths: [
+                    { isAvailable: true },
+                    null,
+                ],
             })
         ).toBeNull()
     })
 })
 
+describe("shouldSkipAnalyzeRecentImagesSchedule", () => {
+    it("keeps an already-scheduled claimable run instead of replacing it", () => {
+        expect(
+            shouldSkipAnalyzeRecentImagesSchedule({
+                now: 10_000,
+                existingNextRunAt: 9_500,
+                requestedNextRunAt: 12_000,
+            })
+        ).toBe(true)
+    })
+
+    it("re-schedules when the existing run is stale in the past", () => {
+        expect(
+            shouldSkipAnalyzeRecentImagesSchedule({
+                now: 10_000,
+                existingNextRunAt: 8_500,
+                requestedNextRunAt: 12_000,
+            })
+        ).toBe(false)
+    })
+
+    it("re-schedules when the new request is earlier than the existing run", () => {
+        expect(
+            shouldSkipAnalyzeRecentImagesSchedule({
+                now: 10_000,
+                existingNextRunAt: 11_000,
+                requestedNextRunAt: 10_500,
+            })
+        ).toBe(false)
+    })
+})
+
+describe("getEffectiveScheduledBackgroundJob", () => {
+    it("prefers the earliest scheduled run when duplicate job rows exist", () => {
+        expect(
+            getEffectiveScheduledBackgroundJob([
+                {
+                    _creationTime: 200,
+                    updatedAt: 2_000,
+                    nextRunAt: 8_000,
+                    scheduledToken: "later",
+                },
+                {
+                    _creationTime: 100,
+                    updatedAt: 1_000,
+                    nextRunAt: 6_000,
+                    scheduledToken: "earlier",
+                },
+            ])
+        ).toMatchObject({
+            nextRunAt: 6_000,
+            scheduledToken: "earlier",
+        })
+    })
+
+    it("breaks same-time schedule ties by the most recent row", () => {
+        expect(
+            getEffectiveScheduledBackgroundJob([
+                {
+                    _creationTime: 100,
+                    updatedAt: 1_000,
+                    nextRunAt: 6_000,
+                    scheduledToken: "older",
+                },
+                {
+                    _creationTime: 200,
+                    updatedAt: 2_000,
+                    nextRunAt: 6_000,
+                    scheduledToken: "newer",
+                },
+            ])
+        ).toMatchObject({
+            nextRunAt: 6_000,
+            scheduledToken: "newer",
+        })
+    })
+})
+
+describe("getLatestBackgroundJobLastRunAt", () => {
+    it("keeps the latest completion timestamp while reconciling duplicates", () => {
+        expect(
+            getLatestBackgroundJobLastRunAt([
+                { lastRunAt: 2_000 },
+                {},
+                { lastRunAt: 5_000 },
+            ])
+        ).toBe(5_000)
+    })
+})
+
 describe("shouldRunRecoveryPromptInference", () => {
     it("runs prompt inference when a prompt exists and no inference was stored", () => {
         expect(