admin/clawhub
1
0
Fork 0
mirror of https://github.com/openclaw/clawhub.git synced 2026-08-14 00:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: bound canonical trending candidate memory (#3332)

Browse Source
This commit is contained in:
Patrick Erichsen
2026-07-30 19:38:00 -07:00
committed by GitHub
parent b7bd53a697
commit 8b0e5b906e
3 changed files with 150 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
convex/canonicalTrending.ts
+37 -7
View File
@@ -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,
convex/lib/canonicalTrending.test.ts
+46
View File
@@ -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", () => {
convex/lib/canonicalTrending.ts
+67 -16
View File
@@ -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<T extends CanonicalTrendingCandidate>(
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<string, number>();
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<T extends CanonicalTrendingCandidate>(
pools: CanonicalTrendingPools<T>,
): CanonicalTrendingPools<T> {
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),
),
};
}