From 8b0e5b906ed1f8a1402feb0cfa1ecf97dcb4b4a9 Mon Sep 17 00:00:00 2001 From: Patrick Erichsen Date: Thu, 30 Jul 2026 19:38:00 -0700 Subject: [PATCH] fix: bound canonical trending candidate memory (#3332) --- convex/canonicalTrending.ts | 44 ++++++++++++--- convex/lib/canonicalTrending.test.ts | 46 +++++++++++++++ convex/lib/canonicalTrending.ts | 83 ++++++++++++++++++++++------ 3 files changed, 150 insertions(+), 23 deletions(-) diff --git a/convex/canonicalTrending.ts b/convex/canonicalTrending.ts index d4f73acb..a1a6d077 100644 --- a/convex/canonicalTrending.ts +++ b/convex/canonicalTrending.ts @@ -4,6 +4,9 @@ import { internal } from "./_generated/api"; import type { Doc, Id } from "./_generated/dataModel"; import { internalAction, internalMutation, internalQuery } from "./_generated/server"; import { + CANONICAL_TRENDING_FIRST_PAGE_SIZE, + CANONICAL_TRENDING_LANE_LIMIT, + CANONICAL_TRENDING_PUBLISHER_CAP, CANONICAL_TRENDING_RANKING_VERSION, CANONICAL_TRENDING_WINDOW_HOURS, blendCanonicalTrendingPools, @@ -14,6 +17,7 @@ import { decodeCanonicalTrendingCursor, encodeCanonicalTrendingCursor, isFreshExternalTrendingRun, + retainTopCanonicalTrendingCandidates, type CanonicalTrendingMaterializationCandidate, } from "./lib/canonicalTrending"; import { forEachCanonicalTrendingSourcePage } from "./lib/canonicalTrendingPagination"; @@ -94,6 +98,11 @@ const operationsValidator = v.object({ functionCalls: v.number(), }); +const LANE_DIVERSITY_RESERVE = { + size: CANONICAL_TRENDING_FIRST_PAGE_SIZE, + publisherCap: CANONICAL_TRENDING_PUBLISHER_CAP, +}; + export const getNativeSourceBatchInternal = internalQuery({ args: { skillIds: v.array(v.id("skills")) }, handler: async (ctx, args) => { @@ -480,8 +489,10 @@ export const materializeInternal = internalAction({ ); finalizeRollingHourlyStats(usageBySkill); - const nativeCandidates: CanonicalTrendingMaterializationCandidate[] = []; + let nativeCandidates: CanonicalTrendingMaterializationCandidate[] = []; + let risingCandidates: CanonicalTrendingMaterializationCandidate[] = []; const nativeSource = { documentsRead: 0, functionCalls: 0 }; + const risingCutoff = startedAt - RISING_MAX_AGE_MS; let pendingNativeSkillIds: Id<"skills">[] = []; const flushNativeSourceBatch = async () => { if (pendingNativeSkillIds.length === 0) return; @@ -497,8 +508,25 @@ export const materializeInternal = internalAction({ 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); + if (!candidate) continue; + nativeCandidates.push(candidate); + if (candidate.createdAt >= risingCutoff) { + risingCandidates.push({ ...candidate, lane: "clawhub-rising" }); + } } + // The fetched batch is capped at 100, so each lane stays within 100 rows of its limit. + nativeCandidates = retainTopCanonicalTrendingCandidates( + nativeCandidates, + "clawhub-trending", + CANONICAL_TRENDING_LANE_LIMIT, + LANE_DIVERSITY_RESERVE, + ); + risingCandidates = retainTopCanonicalTrendingCandidates( + risingCandidates, + "clawhub-rising", + CANONICAL_TRENDING_LANE_LIMIT, + LANE_DIVERSITY_RESERVE, + ); // Each digest is unique by skill, so its rolling totals are no longer needed. for (const skillId of skillIds) usageBySkill.delete(String(skillId)); }; @@ -519,7 +547,7 @@ export const materializeInternal = internalAction({ documentsRead: 0, functionCalls: 0, }; - const externalCandidates: CanonicalTrendingMaterializationCandidate[] = []; + let externalCandidates: CanonicalTrendingMaterializationCandidate[] = []; if ( args.skillsShMode !== "native-only" && getRuntimeRolloutCapabilities().skillsSh.runtimeEnabled @@ -545,6 +573,12 @@ export const materializeInternal = internalAction({ const candidate = buildExternalCanonicalTrendingCandidate(digest); if (candidate) externalCandidates.push(candidate); } + externalCandidates = retainTopCanonicalTrendingCandidates( + externalCandidates, + "skills-sh-trending", + CANONICAL_TRENDING_LANE_LIMIT, + LANE_DIVERSITY_RESERVE, + ); }, ); } @@ -566,10 +600,6 @@ export const materializeInternal = internalAction({ } } - const risingCutoff = startedAt - RISING_MAX_AGE_MS; - const risingCandidates = nativeCandidates - .filter((candidate) => candidate.createdAt >= risingCutoff) - .map((candidate) => ({ ...candidate, lane: "clawhub-rising" as const })); const blended = blendCanonicalTrendingPools({ clawhubTrending: nativeCandidates, clawhubRising: risingCandidates, diff --git a/convex/lib/canonicalTrending.test.ts b/convex/lib/canonicalTrending.test.ts index 3cd45f1c..7f0b3e36 100644 --- a/convex/lib/canonicalTrending.test.ts +++ b/convex/lib/canonicalTrending.test.ts @@ -6,6 +6,7 @@ import { decodeCanonicalTrendingCursor, encodeCanonicalTrendingCursor, isFreshExternalTrendingRun, + retainTopCanonicalTrendingCandidates, sortCanonicalTrendingPools, type CanonicalTrendingCandidate, } from "./canonicalTrending"; @@ -174,6 +175,51 @@ describe("canonical Trending ordering", () => { expect(pools.clawhubRising.map((entry) => entry.identity)).toEqual(["r-new", "r-old"]); expect(pools.skillsShTrending.map((entry) => entry.identity)).toEqual(["s-1", "s-2", "s-3"]); }); + + it("retains only the strongest bounded candidates for a lane", () => { + const retained = retainTopCanonicalTrendingCandidates( + [ + candidate("low", "clawhub-trending", { installs24h: 1 }), + candidate("high", "clawhub-trending", { installs24h: 9 }), + candidate("middle", "clawhub-trending", { installs24h: 4 }), + ], + "clawhub-trending", + 2, + ); + + expect(retained.map((entry) => entry.identity)).toEqual(["high", "middle"]); + }); + + it("reserves lower-ranked publishers needed by the first-page cap", () => { + const retained = retainTopCanonicalTrendingCandidates( + [ + candidate("alpha-1", "clawhub-trending", { + publisherKey: "alpha", + installs24h: 5, + }), + candidate("alpha-2", "clawhub-trending", { + publisherKey: "alpha", + installs24h: 4, + }), + candidate("alpha-3", "clawhub-trending", { + publisherKey: "alpha", + installs24h: 3, + }), + candidate("beta", "clawhub-trending", { publisherKey: "beta", installs24h: 2 }), + candidate("gamma", "clawhub-trending", { publisherKey: "gamma", installs24h: 1 }), + ], + "clawhub-trending", + 2, + { size: 4, publisherCap: 2 }, + ); + + expect(retained.map((entry) => entry.identity)).toEqual([ + "alpha-1", + "alpha-2", + "beta", + "gamma", + ]); + }); }); describe("canonical Trending cursors", () => { diff --git a/convex/lib/canonicalTrending.ts b/convex/lib/canonicalTrending.ts index c3d7a0cb..52b1021e 100644 --- a/convex/lib/canonicalTrending.ts +++ b/convex/lib/canonicalTrending.ts @@ -5,6 +5,7 @@ export const CANONICAL_TRENDING_RANKING_VERSION = "skills-trending-v2"; export const CANONICAL_TRENDING_WINDOW_HOURS = 24; export const CANONICAL_TRENDING_FIRST_PAGE_SIZE = 20; export const CANONICAL_TRENDING_PUBLISHER_CAP = 2; +export const CANONICAL_TRENDING_LANE_LIMIT = 1_000; export function isFreshExternalTrendingRun( run: { runId: string | null; completedAt: number | null }, @@ -309,28 +310,78 @@ function compareIdentity(left: CanonicalTrendingCandidate, right: CanonicalTrend return left.identity.localeCompare(right.identity); } +function compareCanonicalTrendingLaneCandidates( + lane: CanonicalTrendingLane, + left: CanonicalTrendingCandidate, + right: CanonicalTrendingCandidate, +) { + if (lane === "skills-sh-trending") { + return ( + (left.upstreamRank ?? Number.MAX_SAFE_INTEGER) - + (right.upstreamRank ?? Number.MAX_SAFE_INTEGER) || compareIdentity(left, right) + ); + } + return ( + compareNumberDesc(left.installs24h, right.installs24h) || + compareNumberDesc(left.bookmarks24h, right.bookmarks24h) || + compareNumberDesc( + lane === "clawhub-rising" ? left.createdAt : left.updatedAt, + lane === "clawhub-rising" ? right.createdAt : right.updatedAt, + ) || + compareIdentity(left, right) + ); +} + +export function retainTopCanonicalTrendingCandidates( + candidates: readonly T[], + lane: CanonicalTrendingLane, + limit = CANONICAL_TRENDING_LANE_LIMIT, + diversity?: { size: number; publisherCap: number }, +) { + if (!Number.isSafeInteger(limit) || limit < 1) { + throw new Error("Invalid canonical Trending lane limit"); + } + const sorted = [...candidates].sort((left, right) => + compareCanonicalTrendingLaneCandidates(lane, left, right), + ); + const leaders = sorted.slice(0, limit); + if (!diversity) return leaders; + if ( + !Number.isSafeInteger(diversity.size) || + diversity.size < 1 || + !Number.isSafeInteger(diversity.publisherCap) || + diversity.publisherCap < 1 + ) { + throw new Error("Invalid canonical Trending diversity reserve"); + } + const publisherCounts = new Map(); + const reserve: T[] = []; + for (const candidate of sorted) { + const publisherCount = publisherCounts.get(candidate.publisherKey) ?? 0; + if (publisherCount >= diversity.publisherCap) continue; + publisherCounts.set(candidate.publisherKey, publisherCount + 1); + reserve.push(candidate); + if (reserve.length === diversity.size) break; + } + const retainedByIdentity = new Map(leaders.map((candidate) => [candidate.identity, candidate])); + for (const candidate of reserve) retainedByIdentity.set(candidate.identity, candidate); + return [...retainedByIdentity.values()].sort((left, right) => + compareCanonicalTrendingLaneCandidates(lane, left, right), + ); +} + export function sortCanonicalTrendingPools( pools: CanonicalTrendingPools, ): CanonicalTrendingPools { return { - clawhubTrending: [...pools.clawhubTrending].sort( - (left, right) => - compareNumberDesc(left.installs24h, right.installs24h) || - compareNumberDesc(left.bookmarks24h, right.bookmarks24h) || - compareNumberDesc(left.updatedAt, right.updatedAt) || - compareIdentity(left, right), + clawhubTrending: [...pools.clawhubTrending].sort((left, right) => + compareCanonicalTrendingLaneCandidates("clawhub-trending", left, right), ), - clawhubRising: [...pools.clawhubRising].sort( - (left, right) => - compareNumberDesc(left.installs24h, right.installs24h) || - compareNumberDesc(left.bookmarks24h, right.bookmarks24h) || - compareNumberDesc(left.createdAt, right.createdAt) || - compareIdentity(left, right), + clawhubRising: [...pools.clawhubRising].sort((left, right) => + compareCanonicalTrendingLaneCandidates("clawhub-rising", left, right), ), - skillsShTrending: [...pools.skillsShTrending].sort( - (left, right) => - (left.upstreamRank ?? Number.MAX_SAFE_INTEGER) - - (right.upstreamRank ?? Number.MAX_SAFE_INTEGER) || compareIdentity(left, right), + skillsShTrending: [...pools.skillsShTrending].sort((left, right) => + compareCanonicalTrendingLaneCandidates("skills-sh-trending", left, right), ), }; }