fix: stream canonical trending sources (#3329)

convex/_generated/api.d.ts

@@ -64,6 +64,7 @@ import type * as lib_canonicalSkillSearch from "../lib/canonicalSkillSearch.js";
 import type * as lib_canonicalSkillSearchBounds from "../lib/canonicalSkillSearchBounds.js";
 import type * as lib_canonicalSkillSearchResponse from "../lib/canonicalSkillSearchResponse.js";
 import type * as lib_canonicalTrending from "../lib/canonicalTrending.js";
+import type * as lib_canonicalTrendingPagination from "../lib/canonicalTrendingPagination.js";
 import type * as lib_catalogClassification from "../lib/catalogClassification.js";
 import type * as lib_catalogClassifier from "../lib/catalogClassifier.js";
 import type * as lib_changelog from "../lib/changelog.js";
@@ -270,6 +271,7 @@ declare const fullApi: ApiFromModules<{
   "lib/canonicalSkillSearchBounds": typeof lib_canonicalSkillSearchBounds;
   "lib/canonicalSkillSearchResponse": typeof lib_canonicalSkillSearchResponse;
   "lib/canonicalTrending": typeof lib_canonicalTrending;
+  "lib/canonicalTrendingPagination": typeof lib_canonicalTrendingPagination;
   "lib/catalogClassification": typeof lib_catalogClassification;
   "lib/catalogClassifier": typeof lib_catalogClassifier;
   "lib/changelog": typeof lib_changelog;

convex/canonicalTrending.ts

@@ -16,14 +16,19 @@ import {
   isFreshExternalTrendingRun,
   type CanonicalTrendingMaterializationCandidate,
 } from "./lib/canonicalTrending";
+import { forEachCanonicalTrendingSourcePage } from "./lib/canonicalTrendingPagination";
 import { shouldExcludeSkillFromPublicBrowse } from "./lib/publicBrowse";
 import { getRuntimeRolloutCapabilities } from "./lib/rolloutCapabilities";
-import { getCompletedRolling24HourWindow, sumRollingHourlyStats } from "./lib/skillHourlyStats";
+import {
+  accumulateRollingHourlyStats,
+  finalizeRollingHourlyStats,
+  getCompletedRolling24HourWindow,
+  type RollingHourlyStatTotals,
+} from "./lib/skillHourlyStats";
 import { isPublicSkillsShMirrorDigest } from "./lib/skillsShMirrorPublic";
 import { assertTestSeedAllowed } from "./lib/testSeed";
 import { getSkillsShPublicCatalogEnabledHandler } from "./rolloutCapabilities";
 
-const SOURCE_PAGE_SIZE = 250;
 const WRITE_BATCH_SIZE = 100;
 const SNAPSHOT_RETENTION_MS = 48 * 60 * 60 * 1_000;
 const SNAPSHOT_MAX_SERVING_AGE_MS = 2 * 60 * 60 * 1_000;
@@ -88,44 +93,6 @@ const operationsValidator = v.object({
   functionCalls: v.number(),
 });
 
-type SourcePage<T> = {
-  page: T[];
-  isDone: boolean;
-  continueCursor: string;
-  documentsRead: number;
-};
-
-type CollectedSource<T> = {
-  rows: T[];
-  documentsRead: number;
-  functionCalls: number;
-};
-
-async function collectSourcePages(
-  ctx: { runQuery: (ref: never, args: never) => Promise<unknown> },
-  ref: unknown,
-  args: Record<string, unknown> = {},
-) {
-  const rows: unknown[] = [];
-  let cursor: string | null = null;
-  let documentsRead = 0;
-  let functionCalls = 0;
-  do {
-    const result = (await ctx.runQuery(
-      ref as never,
-      {
-        ...args,
-        paginationOpts: { cursor, numItems: SOURCE_PAGE_SIZE },
-      } as never,
-    )) as SourcePage<unknown>;
-    rows.push(...result.page);
-    documentsRead += result.documentsRead;
-    functionCalls += 1;
-    cursor = result.isDone ? null : result.continueCursor;
-  } while (cursor);
-  return { rows, documentsRead, functionCalls };
-}
-
 export const getNativeSourcePageInternal = internalQuery({
   args: { paginationOpts: paginationOptsValidator },
   handler: async (ctx, args) => {
@@ -491,11 +458,8 @@ export const materializeInternal = internalAction({
         return { status: "unavailable" as const, reason: "hourly-stats-not-ready" as const };
       }
 
-      const nativeSource = (await collectSourcePages(
-        ctx,
-        internalRefs.canonicalTrending.getNativeSourcePageInternal,
-      )) as CollectedSource<Doc<"skillSearchDigest">>;
-      const hourlySource = (await collectSourcePages(
+      const usageBySkill: RollingHourlyStatTotals = new Map();
+      const hourlySource = await forEachCanonicalTrendingSourcePage(
         ctx,
         internalRefs.canonicalTrending.getHourlySourcePageInternal,
         {
@@ -503,18 +467,35 @@ export const materializeInternal = internalAction({
           endHour: hourlyWindow.endHour,
           maxGeneration: hourlyWindow.sealedGeneration,
         },
-      )) as CollectedSource<Doc<"skillHourlyStats">>;
+        (page) => accumulateRollingHourlyStats(usageBySkill, page as Doc<"skillHourlyStats">[]),
+      );
+      finalizeRollingHourlyStats(usageBySkill);
+
+      const nativeCandidates: CanonicalTrendingMaterializationCandidate[] = [];
+      const nativeSource = await forEachCanonicalTrendingSourcePage(
+        ctx,
+        internalRefs.canonicalTrending.getNativeSourcePageInternal,
+        {},
+        (page) => {
+          for (const digest of page as Doc<"skillSearchDigest">[]) {
+            const usage = usageBySkill.get(String(digest.skillId));
+            if (!usage || usage.downloads + usage.installs + usage.bookmarks <= 0) continue;
+            const candidate = buildNativeCanonicalTrendingCandidate(digest, usage);
+            if (candidate) nativeCandidates.push(candidate);
+          }
+        },
+      );
       type TrendingRun = {
         runId: Doc<"skillsShMirrorRuns">["_id"] | null;
         completedAt: number | null;
         documentsRead: number;
       };
       let latestTrendingRun: TrendingRun | null = null;
-      let externalSource: CollectedSource<Doc<"skillsShMirrorDigests">> = {
-        rows: [],
+      let externalSource = {
         documentsRead: 0,
         functionCalls: 0,
       };
+      const externalCandidates: CanonicalTrendingMaterializationCandidate[] = [];
       if (
         args.skillsShMode !== "native-only" &&
         getRuntimeRolloutCapabilities().skillsSh.runtimeEnabled
@@ -527,14 +508,21 @@ export const materializeInternal = internalAction({
         functionCalls += 1;
         if (isFreshExternalTrendingRun(candidateRun, startedAt, EXTERNAL_SOURCE_MAX_AGE_MS)) {
           latestTrendingRun = candidateRun;
-          externalSource = (await collectSourcePages(
+          externalSource = await forEachCanonicalTrendingSourcePage(
             ctx,
             internalRefs.canonicalTrending.getExternalSourcePageInternal,
             {
               activationLockToken: args.activationLockToken,
               allowHiddenProof: args.proofSnapshotId !== undefined,
             },
-          )) as CollectedSource<Doc<"skillsShMirrorDigests">>;
+            (page) => {
+              for (const digest of page as Doc<"skillsShMirrorDigests">[]) {
+                if (digest.trendingObservedRunId !== latestTrendingRun?.runId) continue;
+                const candidate = buildExternalCanonicalTrendingCandidate(digest);
+                if (candidate) externalCandidates.push(candidate);
+              }
+            },
+          );
         }
       }
       documentsRead +=
@@ -554,26 +542,10 @@ export const materializeInternal = internalAction({
         }
       }
 
-      const usageBySkill = sumRollingHourlyStats(hourlySource.rows);
-      const nativeCandidates = nativeSource.rows
-        .map((digest) => {
-          const usage = usageBySkill.get(String(digest.skillId));
-          if (!usage || usage.downloads + usage.installs + usage.bookmarks <= 0) return null;
-          return buildNativeCanonicalTrendingCandidate(digest, usage);
-        })
-        .filter(
-          (candidate): candidate is CanonicalTrendingMaterializationCandidate => candidate !== null,
-        );
       const risingCutoff = startedAt - RISING_MAX_AGE_MS;
       const risingCandidates = nativeCandidates
         .filter((candidate) => candidate.createdAt >= risingCutoff)
         .map((candidate) => ({ ...candidate, lane: "clawhub-rising" as const }));
-      const externalCandidates = externalSource.rows
-        .filter((digest) => digest.trendingObservedRunId === latestTrendingRun?.runId)
-        .map(buildExternalCanonicalTrendingCandidate)
-        .filter(
-          (candidate): candidate is CanonicalTrendingMaterializationCandidate => candidate !== null,
-        );
       const blended = blendCanonicalTrendingPools({
         clawhubTrending: nativeCandidates,
         clawhubRising: risingCandidates,

convex/lib/canonicalTrendingPagination.test.ts

@@ -0,0 +1,36 @@
+import { describe, expect, it, vi } from "vitest";
+import type { ActionCtx } from "../_generated/server";
+import { forEachCanonicalTrendingSourcePage } from "./canonicalTrendingPagination";
+
+describe("canonical Trending source pagination", () => {
+  it("consumes each page without returning the accumulated source rows", async () => {
+    const pages = [
+      { page: [{ id: "one" }, { id: "two" }], documentsRead: 2 },
+      { page: [{ id: "three" }], documentsRead: 1 },
+    ];
+    const runQuery = vi.fn(
+      async (_ref: never, args: { paginationOpts: { cursor: string | null } }) => {
+        const index = args.paginationOpts.cursor === null ? 0 : 1;
+        return {
+          ...pages[index],
+          isDone: index === pages.length - 1,
+          continueCursor: index === pages.length - 1 ? "" : "next-page",
+        };
+      },
+    );
+    const visited: string[][] = [];
+
+    const result = await forEachCanonicalTrendingSourcePage(
+      { runQuery: runQuery as unknown as ActionCtx["runQuery"] },
+      Symbol("source"),
+      {},
+      (page) => {
+        visited.push((page as Array<{ id: string }>).map((row) => row.id));
+      },
+    );
+
+    expect(visited).toEqual([["one", "two"], ["three"]]);
+    expect(result).toEqual({ documentsRead: 3, functionCalls: 2 });
+    expect(result).not.toHaveProperty("rows");
+  });
+});

convex/lib/canonicalTrendingPagination.ts

@@ -0,0 +1,34 @@
+import type { ActionCtx } from "../_generated/server";
+
+type SourcePage = {
+  page: unknown[];
+  isDone: boolean;
+  continueCursor: string;
+  documentsRead: number;
+};
+
+export async function forEachCanonicalTrendingSourcePage(
+  ctx: Pick<ActionCtx, "runQuery">,
+  ref: unknown,
+  args: Record<string, unknown>,
+  visitPage: (page: unknown[]) => void | Promise<void>,
+  pageSize = 250,
+) {
+  let cursor: string | null = null;
+  let documentsRead = 0;
+  let functionCalls = 0;
+  do {
+    const result = (await ctx.runQuery(
+      ref as never,
+      {
+        ...args,
+        paginationOpts: { cursor, numItems: pageSize },
+      } as never,
+    )) as SourcePage;
+    await visitPage(result.page);
+    documentsRead += result.documentsRead;
+    functionCalls += 1;
+    cursor = result.isDone ? null : result.continueCursor;
+  } while (cursor);
+  return { documentsRead, functionCalls };
+}

convex/lib/skillHourlyStats.test.ts

@@ -6,8 +6,10 @@ import { internal } from "../_generated/api";
 import schema from "../schema";
 import {
   HOUR_MS,
+  accumulateRollingHourlyStats,
   bumpHistoricalHourlySkillStats,
   bumpLiveHourlySkillStats,
+  finalizeRollingHourlyStats,
   getHistoricalEventHourlyDelta,
   getHistoricalStarHourlyDelta,
   getCompletedRolling24HourWindow,
@@ -61,6 +63,42 @@ describe("rolling 24-hour skill metrics", () => {
     );
   });
 
+  it("preserves corrections across streamed page boundaries before clamping", () => {
+    const totals = new Map();
+    accumulateRollingHourlyStats(totals, [
+      {
+        skillId: "skills:one",
+        downloads: -4,
+        installs: -3,
+        bookmarks: -2,
+        updatedAt: 10,
+      },
+    ]);
+    accumulateRollingHourlyStats(totals, [
+      {
+        skillId: "skills:one",
+        downloads: 6,
+        installs: 5,
+        bookmarks: 4,
+        updatedAt: 20,
+      },
+    ]);
+
+    expect(finalizeRollingHourlyStats(totals)).toEqual(
+      new Map([
+        [
+          "skills:one",
+          {
+            downloads: 2,
+            installs: 2,
+            bookmarks: 2,
+            updatedAt: 20,
+          },
+        ],
+      ]),
+    );
+  });
+
   it("keeps historical seed counts separate from concurrent live deltas", async () => {
     const t = convexTest(schema, modules);
     const hourAt = 90 * HOUR_MS;

convex/lib/skillHourlyStats.ts

@@ -27,7 +27,7 @@ export function getCompletedRolling24HourWindow(now: number) {
   };
 }
 
-type HourlySkillStat = {
+export type HourlySkillStat = {
   skillId: string;
   downloads: number;
   installs: number;
@@ -35,11 +35,15 @@ type HourlySkillStat = {
   updatedAt: number;
 };
 
-export function sumRollingHourlyStats(rows: readonly HourlySkillStat[]) {
-  const totals = new Map<
-    string,
-    { downloads: number; installs: number; bookmarks: number; updatedAt: number }
-  >();
+export type RollingHourlyStatTotals = Map<
+  string,
+  { downloads: number; installs: number; bookmarks: number; updatedAt: number }
+>;
+
+export function accumulateRollingHourlyStats(
+  totals: RollingHourlyStatTotals,
+  rows: readonly HourlySkillStat[],
+) {
   for (const row of rows) {
     const current = totals.get(row.skillId) ?? {
       downloads: 0,
@@ -53,6 +57,9 @@ export function sumRollingHourlyStats(rows: readonly HourlySkillStat[]) {
     current.updatedAt = Math.max(current.updatedAt, row.updatedAt);
     totals.set(row.skillId, current);
   }
+}
+
+export function finalizeRollingHourlyStats(totals: RollingHourlyStatTotals) {
   for (const total of totals.values()) {
     total.downloads = Math.max(0, total.downloads);
     total.installs = Math.max(0, total.installs);
@@ -61,6 +68,12 @@ export function sumRollingHourlyStats(rows: readonly HourlySkillStat[]) {
   return totals;
 }
 
+export function sumRollingHourlyStats(rows: readonly HourlySkillStat[]) {
+  const totals: RollingHourlyStatTotals = new Map();
+  accumulateRollingHourlyStats(totals, rows);
+  return finalizeRollingHourlyStats(totals);
+}
+
 type HourlyStatDeltas = {
   skillId: Id<"skills">;
   occurredAt: number;