From e986ac3b022c4f6fe031aa642edc1c54c580e2d4 Mon Sep 17 00:00:00 2001 From: Patrick Erichsen Date: Thu, 30 Jul 2026 17:59:33 -0700 Subject: [PATCH] fix: stream canonical trending sources (#3329) --- convex/_generated/api.d.ts | 2 + convex/canonicalTrending.ts | 104 +++++++----------- .../lib/canonicalTrendingPagination.test.ts | 36 ++++++ convex/lib/canonicalTrendingPagination.ts | 34 ++++++ convex/lib/skillHourlyStats.test.ts | 38 +++++++ convex/lib/skillHourlyStats.ts | 25 ++++- 6 files changed, 167 insertions(+), 72 deletions(-) create mode 100644 convex/lib/canonicalTrendingPagination.test.ts create mode 100644 convex/lib/canonicalTrendingPagination.ts diff --git a/convex/_generated/api.d.ts b/convex/_generated/api.d.ts index 6671966e..b4e246cf 100644 --- a/convex/_generated/api.d.ts +++ b/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; diff --git a/convex/canonicalTrending.ts b/convex/canonicalTrending.ts index e3d06ad5..afc73cc7 100644 --- a/convex/canonicalTrending.ts +++ b/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 = { - page: T[]; - isDone: boolean; - continueCursor: string; - documentsRead: number; -}; - -type CollectedSource = { - rows: T[]; - documentsRead: number; - functionCalls: number; -}; - -async function collectSourcePages( - ctx: { runQuery: (ref: never, args: never) => Promise }, - ref: unknown, - args: Record = {}, -) { - 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; - 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>; - 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>; + (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> = { - 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>; + (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, diff --git a/convex/lib/canonicalTrendingPagination.test.ts b/convex/lib/canonicalTrendingPagination.test.ts new file mode 100644 index 00000000..d54d8424 --- /dev/null +++ b/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"); + }); +}); diff --git a/convex/lib/canonicalTrendingPagination.ts b/convex/lib/canonicalTrendingPagination.ts new file mode 100644 index 00000000..79587dfc --- /dev/null +++ b/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, + ref: unknown, + args: Record, + visitPage: (page: unknown[]) => void | Promise, + 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 }; +} diff --git a/convex/lib/skillHourlyStats.test.ts b/convex/lib/skillHourlyStats.test.ts index b954e04f..326907b5 100644 --- a/convex/lib/skillHourlyStats.test.ts +++ b/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; diff --git a/convex/lib/skillHourlyStats.ts b/convex/lib/skillHourlyStats.ts index 2dd157bb..134353fa 100644 --- a/convex/lib/skillHourlyStats.ts +++ b/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;