From 455f4ea19c85c994f0af57cbbf3fdd0773e65f6e Mon Sep 17 00:00:00 2001 From: Patrick Erichsen Date: Mon, 15 Jun 2026 17:48:58 -0700 Subject: [PATCH] fix: serialize skill stat event drain Serialize the skill document stat drain behind a short-lived Convex lease and add admin kick/status helpers for manual backlog recovery. --- bun.lock | 3 +- convex/llmEval.ts | 9 +- convex/schema.ts | 12 ++ convex/skillStatEvents.test.ts | 43 ++++- convex/skillStatEvents.ts | 328 +++++++++++++++++++++++++++++++-- convex/skills.ts | 4 +- convex/vt.ts | 17 +- package.json | 1 + 8 files changed, 379 insertions(+), 38 deletions(-) diff --git a/bun.lock b/bun.lock index 76634a01..c87ade8b 100644 --- a/bun.lock +++ b/bun.lock @@ -148,6 +148,7 @@ }, }, "overrides": { + "ast-v8-to-istanbul": "1.0.4", "dompurify": "3.4.10", "next": "16.2.6", "postcss": "8.5.12", @@ -960,7 +961,7 @@ "assertion-error": ["assertion-error@2.0.1", "", {}, "sha512-Izi8RQcffqCeNVgFigKli1ssklIbpHnCYc6AknXGYoB6grJqyeby7jv12JUQgmTAnIDnbck1uxksT4dzN3PWBA=="], - "ast-v8-to-istanbul": ["ast-v8-to-istanbul@1.0.3", "", { "dependencies": { "@jridgewell/trace-mapping": "^0.3.31", "estree-walker": "^3.0.3", "js-tokens": "^10.0.0" } }, "sha512-jCMQ6ZylLPudp0CDfBmQBZUsrh1/8psbmu9ibeVWKuHWD0YrH9YABwlKu5kVEFoT0GCQQW9Z/SxfuEbbkGQCRg=="], + "ast-v8-to-istanbul": ["ast-v8-to-istanbul@1.0.4", "", { "dependencies": { "@jridgewell/trace-mapping": "^0.3.31", "estree-walker": "^3.0.3", "js-tokens": "^10.0.0" } }, "sha512-0bC0/4bTSrnwdhU3IsZDwEdojvuPrSg59OYZfKsLRtJZ0u8VBx9DebfqqG8bRdCC0I7vjgxmPi41P0lpkhJHtA=="], "atomically": ["atomically@2.1.1", "", { "dependencies": { "stubborn-fs": "^2.0.0", "when-exit": "^2.1.4" } }, "sha512-P4w9o2dqARji6P7MHprklbfiArZAWvo07yW7qs3pdljb3BWr12FIB7W+p0zJiuiVsUpRO0iZn1kFFcpPegg0tQ=="], diff --git a/convex/llmEval.ts b/convex/llmEval.ts index dd78c8e5..6af71f1e 100644 --- a/convex/llmEval.ts +++ b/convex/llmEval.ts @@ -249,9 +249,12 @@ export const evaluateWithLlm = internalAction({ return; } - const fingerprintEntries = await ctx.runQuery(internal.skills.listVersionFingerprintsInternal, { - skillVersionId: version._id, - }); + const fingerprintEntries = (await ctx.runQuery( + internal.skills.listVersionFingerprintsInternal, + { + skillVersionId: version._id, + }, + )) as Array<{ fingerprint: string; kind?: "source" | "generated-bundle" }>; const generatedBundleFingerprints = fingerprintEntries .filter((entry) => entry.kind === "generated-bundle") .map((entry) => entry.fingerprint); diff --git a/convex/schema.ts b/convex/schema.ts index e116870e..dce7bbc6 100644 --- a/convex/schema.ts +++ b/convex/schema.ts @@ -1931,6 +1931,17 @@ const skillStatUpdateCursors = defineTable({ updatedAt: v.number(), }).index("by_key", ["key"]); +const skillStatDocSyncLeases = defineTable({ + key: v.string(), + leaseOwner: v.string(), + leaseExpiresAt: v.number(), + updatedAt: v.number(), + lastStartedAt: v.optional(v.number()), + lastFinishedAt: v.optional(v.number()), + lastProcessedAt: v.optional(v.number()), + lastProcessedCount: v.optional(v.number()), +}).index("by_key", ["key"]); + const comments = defineTable({ skillId: v.id("skills"), userId: v.id("users"), @@ -2597,6 +2608,7 @@ export default defineSchema({ globalStats, skillStatEvents, skillStatUpdateCursors, + skillStatDocSyncLeases, comments, commentReports, skillReports, diff --git a/convex/skillStatEvents.test.ts b/convex/skillStatEvents.test.ts index 0d0c3e56..9e554e80 100644 --- a/convex/skillStatEvents.test.ts +++ b/convex/skillStatEvents.test.ts @@ -10,17 +10,23 @@ vi.mock("./functions", () => ({ vi.mock("./_generated/api", () => ({ internal: { skillStatEvents: { + claimSkillStatDocSyncLeaseInternal: Symbol("claimSkillStatDocSyncLeaseInternal"), + processSkillStatEventBatchInternal: Symbol("processSkillStatEventBatchInternal"), processSkillStatEventsAction: Symbol("processSkillStatEventsAction"), processSkillStatEventsInternal: Symbol("processSkillStatEventsInternal"), + releaseSkillStatDocSyncLeaseInternal: Symbol("releaseSkillStatDocSyncLeaseInternal"), }, }, })); -const { processSkillStatEventsInternal } = await import("./skillStatEvents"); +const { processSkillStatEventBatchInternal } = await import("./skillStatEvents"); -const processSkillStatEventsInternalHandler = ( - processSkillStatEventsInternal as unknown as { - _handler: (ctx: unknown, args: { batchSize?: number }) => Promise<{ processed: number }>; +const processSkillStatEventBatchInternalHandler = ( + processSkillStatEventBatchInternal as unknown as { + _handler: ( + ctx: unknown, + args: { batchSize?: number; leaseOwner: string }, + ) => Promise<{ processed: number }>; } )._handler; @@ -54,11 +60,25 @@ describe("skill stat events - comment delta handling", () => { }, }; const patch = vi.fn(); + const lease = { + _id: "skillStatDocSyncLeases:1", + key: "skill_doc_stat_sync", + leaseOwner: "test-lease", + leaseExpiresAt: Date.now() + 60_000, + updatedAt: Date.now(), + }; const ctx = { db: { get: vi.fn(async (id: string) => (id === "skills:1" ? skill : null)), patch, query: vi.fn((table: string) => { + if (table === "skillStatDocSyncLeases") { + return { + withIndex: () => ({ + unique: async () => lease, + }), + }; + } if (table !== "skillStatEvents") throw new Error(`unexpected table ${table}`); return { withIndex: () => ({ @@ -70,15 +90,26 @@ describe("skill stat events - comment delta handling", () => { scheduler: { runAfter: vi.fn() }, }; - await expect(processSkillStatEventsInternalHandler(ctx, { batchSize: 10 })).resolves.toEqual({ + await expect( + processSkillStatEventBatchInternalHandler(ctx, { + batchSize: 10, + leaseOwner: "test-lease", + }), + ).resolves.toEqual({ + hasMore: false, processed: 1, + skillsUpdated: 0, }); - expect(patch).toHaveBeenCalledTimes(1); + expect(patch).toHaveBeenCalledTimes(2); expect(patch).toHaveBeenCalledWith( "skillStatEvents:star", expect.objectContaining({ processedAt: expect.any(Number) }), ); + expect(patch).toHaveBeenCalledWith( + "skillStatDocSyncLeases:1", + expect.objectContaining({ lastProcessedCount: 1 }), + ); }); it("aggregates comment and uncomment events into net deltas", () => { diff --git a/convex/skillStatEvents.ts b/convex/skillStatEvents.ts index 2ec34b29..784650bb 100644 --- a/convex/skillStatEvents.ts +++ b/convex/skillStatEvents.ts @@ -178,34 +178,178 @@ function aggregateEvents(events: Doc<"skillStatEvents">[]): AggregatedDeltas { return result; } +const DOC_SYNC_LEASE_KEY = "skill_doc_stat_sync"; +const DOC_SYNC_LEASE_MS = 2 * 60 * 1_000; +const DEFAULT_DOC_SYNC_BATCH_SIZE = 100; +const MAX_DOC_SYNC_BATCH_SIZE = 100; +const DEFAULT_DOC_SYNC_MAX_BATCHES = 20; +const MAX_DOC_SYNC_MAX_BATCHES = 100; + +type ClaimSkillStatDocSyncLeaseResult = + | { + acquired: true; + leaseOwner: string; + leaseExpiresAt: number; + now: number; + } + | { + acquired: false; + leaseOwner: string; + leaseExpiresAt: number; + now: number; + }; + +type SkillStatDocSyncBatchResult = { + processed: number; + skillsUpdated: number; + hasMore: boolean; + skipped?: "lease_lost"; +}; + +type SkillStatDocSyncActionResult = + | { + acquired: false; + processed: number; + skillsUpdated: number; + scheduledContinuation: false; + leaseExpiresAt: number; + now: number; + } + | { + acquired: true; + processed: number; + skillsUpdated: number; + batches: number; + stoppedReason: "empty" | "max_batches" | "lease_lost"; + scheduledContinuation: boolean; + }; + +function clampInt(value: number, min: number, max: number) { + if (!Number.isFinite(value)) return min; + return Math.max(min, Math.min(Math.floor(value), max)); +} + +function normalizeDocSyncBatchSize(batchSize: number | undefined) { + return clampInt(batchSize ?? DEFAULT_DOC_SYNC_BATCH_SIZE, 1, MAX_DOC_SYNC_BATCH_SIZE); +} + +function normalizeDocSyncMaxBatches(maxBatches: number | undefined) { + return clampInt(maxBatches ?? DEFAULT_DOC_SYNC_MAX_BATCHES, 1, MAX_DOC_SYNC_MAX_BATCHES); +} + +export const claimSkillStatDocSyncLeaseInternal = internalMutation({ + args: { leaseMs: v.optional(v.number()) }, + handler: async (ctx, args): Promise => { + const now = Date.now(); + const leaseMs = clampInt(args.leaseMs ?? DOC_SYNC_LEASE_MS, 30_000, 10 * 60 * 1_000); + const existing = await ctx.db + .query("skillStatDocSyncLeases") + .withIndex("by_key", (q) => q.eq("key", DOC_SYNC_LEASE_KEY)) + .unique(); + + if (existing && existing.leaseExpiresAt > now) { + return { + acquired: false as const, + leaseOwner: existing.leaseOwner, + leaseExpiresAt: existing.leaseExpiresAt, + now, + }; + } + + const leaseOwner = `${now}`; + const patch = { + leaseOwner, + leaseExpiresAt: now + leaseMs, + updatedAt: now, + lastStartedAt: now, + }; + + if (existing) { + await ctx.db.patch(existing._id, patch); + } else { + await ctx.db.insert("skillStatDocSyncLeases", { + key: DOC_SYNC_LEASE_KEY, + ...patch, + }); + } + + return { + acquired: true as const, + leaseOwner, + leaseExpiresAt: now + leaseMs, + now, + }; + }, +}); + +export const releaseSkillStatDocSyncLeaseInternal = internalMutation({ + args: { + leaseOwner: v.string(), + processed: v.optional(v.number()), + }, + handler: async (ctx, args) => { + const now = Date.now(); + const lease = await ctx.db + .query("skillStatDocSyncLeases") + .withIndex("by_key", (q) => q.eq("key", DOC_SYNC_LEASE_KEY)) + .unique(); + + if (!lease || lease.leaseOwner !== args.leaseOwner) { + return { released: false as const }; + } + + await ctx.db.patch(lease._id, { + leaseExpiresAt: now, + updatedAt: now, + lastFinishedAt: now, + lastProcessedCount: args.processed ?? lease.lastProcessedCount, + }); + + return { released: true as const }; + }, +}); + /** * Process a batch of unprocessed stat events. * - * Called by the 6-hour cron to sync stats to skill docs. Processes up to batchSize events (default 500). - * If the batch is full, schedules an immediate follow-up run to drain the queue. + * Called by the leased action drain to sync stats to skill docs. Processes up to + * batchSize events. The committed lease check keeps overlapping cron/manual + * kicks from doing the same heavy work before Convex's OCC retry machinery + * chooses a winner. * * Processing steps: - * 1. Query unprocessed events (processedAt is undefined) - * 2. Group events by skillId to minimize skill document fetches - * 3. For each skill: + * 1. Verify the committed lease owner + * 2. Query unprocessed events (processedAt is undefined) + * 3. Group events by skillId to minimize skill document fetches + * 4. For each skill: * a. Fetch the skill document once * b. Aggregate all events for this skill into net deltas * c. Apply deltas to skill stats (downloads, stars, installs) - * d. Update daily stats for trending (using original event timestamps) - * e. Mark all events as processed - * 4. If batch was full, schedule another run immediately + * d. Mark all events as processed * * Aggregation levels: * - Level 1: Batch of 100 events from the queue * - Level 2: Group by skillId (e.g., 100 events → 30 unique skills) * - Level 3: Aggregate events per skill (e.g., 5 events → 1 skill update) - * - Level 4: Daily stats may coalesce (e.g., 3 downloads same day → 1 upsert) */ -export const processSkillStatEventsInternal = internalMutation({ - args: { batchSize: v.optional(v.number()) }, - handler: async (ctx, args) => { - const batchSize = Math.max(1, Math.min(args.batchSize ?? 100, 100)); +export const processSkillStatEventBatchInternal = internalMutation({ + args: { batchSize: v.optional(v.number()), leaseOwner: v.string() }, + handler: async (ctx, args): Promise => { + const batchSize = normalizeDocSyncBatchSize(args.batchSize); const now = Date.now(); + const lease = await ctx.db + .query("skillStatDocSyncLeases") + .withIndex("by_key", (q) => q.eq("key", DOC_SYNC_LEASE_KEY)) + .unique(); + + if (!lease || lease.leaseOwner !== args.leaseOwner || lease.leaseExpiresAt <= now) { + return { + processed: 0, + skillsUpdated: 0, + hasMore: false, + skipped: "lease_lost" as const, + }; + } // Level 1: Fetch a batch of unprocessed events const events = await ctx.db @@ -214,7 +358,7 @@ export const processSkillStatEventsInternal = internalMutation({ .take(batchSize); if (events.length === 0) { - return { processed: 0 }; + return { processed: 0, skillsUpdated: 0, hasMore: false }; } // Level 2: Group events by skillId to minimize database reads @@ -228,6 +372,7 @@ export const processSkillStatEventsInternal = internalMutation({ } // Process each skill's events + let skillsUpdated = 0; for (const [skillId, skillEvents] of eventsBySkill) { const skill = await ctx.db.get(skillId); @@ -262,6 +407,7 @@ export const processSkillStatEventsInternal = internalMutation({ // skill's position in the by_active_updated index. await ctx.db.patch(skill._id, patch); await adjustUserSkillStatsForSkillChange(ctx, skill, { ...skill, ...patch }); + skillsUpdated += 1; } // NOTE: Daily stats (skillDailyStats) are written by the 15-minute @@ -273,16 +419,160 @@ export const processSkillStatEventsInternal = internalMutation({ } } - // If we hit the batch limit, there may be more events waiting. - // Schedule an immediate follow-up run to drain the queue. - // This ensures high-volume periods don't create a backlog. - if (events.length === batchSize) { + await ctx.db.patch(lease._id, { + leaseExpiresAt: now + DOC_SYNC_LEASE_MS, + updatedAt: now, + lastProcessedAt: now, + lastProcessedCount: events.length, + }); + + return { + processed: events.length, + skillsUpdated, + hasMore: events.length === batchSize, + }; + }, +}); + +/** + * Leased skill-document stat sync drain. + * + * This action is the cron/manual entrypoint. It commits a lease before doing + * batch work so concurrent scheduled runs skip quickly instead of processing + * the same first unprocessed rows and relying on OCC to throw one away. + */ +export const processSkillStatEventsInternal: ReturnType = internalAction({ + args: { + batchSize: v.optional(v.number()), + maxBatches: v.optional(v.number()), + }, + handler: async (ctx, args): Promise => { + const batchSize = normalizeDocSyncBatchSize(args.batchSize); + const maxBatches = normalizeDocSyncMaxBatches(args.maxBatches); + const claim: ClaimSkillStatDocSyncLeaseResult = await ctx.runMutation( + internal.skillStatEvents.claimSkillStatDocSyncLeaseInternal, + { + leaseMs: DOC_SYNC_LEASE_MS, + }, + ); + + if (!claim.acquired) { + return { + acquired: false as const, + processed: 0, + skillsUpdated: 0, + scheduledContinuation: false, + leaseExpiresAt: claim.leaseExpiresAt, + now: claim.now, + }; + } + + let processed = 0; + let skillsUpdated = 0; + let batches = 0; + let hasMore = false; + let stoppedReason: "empty" | "max_batches" | "lease_lost" = "empty"; + + for (let index = 0; index < maxBatches; index += 1) { + const batch: SkillStatDocSyncBatchResult = await ctx.runMutation( + internal.skillStatEvents.processSkillStatEventBatchInternal, + { + batchSize, + leaseOwner: claim.leaseOwner, + }, + ); + + if (batch.skipped === "lease_lost") { + stoppedReason = "lease_lost"; + hasMore = false; + break; + } + + batches += 1; + processed += batch.processed; + skillsUpdated += batch.skillsUpdated; + hasMore = batch.hasMore; + + if (!batch.hasMore) { + stoppedReason = "empty"; + break; + } + + stoppedReason = "max_batches"; + } + + await ctx.runMutation(internal.skillStatEvents.releaseSkillStatDocSyncLeaseInternal, { + leaseOwner: claim.leaseOwner, + processed, + }); + + if (hasMore && stoppedReason === "max_batches") { await ctx.scheduler.runAfter(0, internal.skillStatEvents.processSkillStatEventsInternal, { batchSize, + maxBatches, }); } - return { processed: events.length }; + return { + acquired: true as const, + processed, + skillsUpdated, + batches, + stoppedReason, + scheduledContinuation: hasMore && stoppedReason === "max_batches", + }; + }, +}); + +export const kickSkillStatDocSyncInternal = internalMutation({ + args: { + batchSize: v.optional(v.number()), + maxBatches: v.optional(v.number()), + }, + handler: async (ctx, args) => { + const batchSize = normalizeDocSyncBatchSize(args.batchSize); + const maxBatches = normalizeDocSyncMaxBatches(args.maxBatches); + await ctx.scheduler.runAfter(0, internal.skillStatEvents.processSkillStatEventsInternal, { + batchSize, + maxBatches, + }); + return { ok: true as const, batchSize, maxBatches }; + }, +}); + +export const getSkillStatDocSyncStatusInternal = internalQuery({ + args: { sampleLimit: v.optional(v.number()) }, + handler: async (ctx, args) => { + const now = Date.now(); + const sampleLimit = clampInt(args.sampleLimit ?? 1_000, 1, 10_000); + const events = await ctx.db + .query("skillStatEvents") + .withIndex("by_unprocessed", (q) => q.eq("processedAt", undefined)) + .take(sampleLimit); + const lease = await ctx.db + .query("skillStatDocSyncLeases") + .withIndex("by_key", (q) => q.eq("key", DOC_SYNC_LEASE_KEY)) + .unique(); + + return { + hasPending: events.length > 0, + samplePendingCount: events.length, + sampleLimit, + sampledToLimit: events.length === sampleLimit, + oldestPendingAt: events[0]?.occurredAt, + newestPendingAt: events[events.length - 1]?.occurredAt, + lease: lease + ? { + active: lease.leaseExpiresAt > now, + leaseExpiresAt: lease.leaseExpiresAt, + lastStartedAt: lease.lastStartedAt, + lastFinishedAt: lease.lastFinishedAt, + lastProcessedAt: lease.lastProcessedAt, + lastProcessedCount: lease.lastProcessedCount, + } + : null, + now, + }; }, }); diff --git a/convex/skills.ts b/convex/skills.ts index 7292d9ce..895c9459 100644 --- a/convex/skills.ts +++ b/convex/skills.ts @@ -7442,12 +7442,12 @@ export const scanSkillVersionStaticallyInternal: ReturnType; const generatedBundleFingerprints = fingerprintEntries .filter((entry) => entry.kind === "generated-bundle") .map((entry) => entry.fingerprint); diff --git a/convex/vt.ts b/convex/vt.ts index 3c691b2e..1ae3b1bd 100644 --- a/convex/vt.ts +++ b/convex/vt.ts @@ -493,9 +493,9 @@ export const scanWithVirusTotal = internalAction({ } // Get the version details and files - const version = await ctx.runQuery(internal.skills.getVersionByIdInternal, { + const version = (await ctx.runQuery(internal.skills.getVersionByIdInternal, { versionId: args.versionId, - }); + })) as Doc<"skillVersions"> | null; if (!version) { console.error(`Version ${args.versionId} not found for scanning`); @@ -503,17 +503,20 @@ export const scanWithVirusTotal = internalAction({ } // Fetch skill info for _meta.json - const skill = await ctx.runQuery(internal.skills.getSkillByIdInternal, { + const skill = (await ctx.runQuery(internal.skills.getSkillByIdInternal, { skillId: version.skillId, - }); + })) as Doc<"skills"> | null; if (!skill) { console.error(`Skill ${version.skillId} not found for scanning`); return; } - const fingerprintEntries = await ctx.runQuery(internal.skills.listVersionFingerprintsInternal, { - skillVersionId: version._id, - }); + const fingerprintEntries = (await ctx.runQuery( + internal.skills.listVersionFingerprintsInternal, + { + skillVersionId: version._id, + }, + )) as Array<{ fingerprint: string; kind?: "source" | "generated-bundle" }>; const generatedBundleFingerprints = fingerprintEntries .filter((entry) => entry.kind === "generated-bundle") .map((entry) => entry.fingerprint); diff --git a/package.json b/package.json index 31808d57..67d1b837 100644 --- a/package.json +++ b/package.json @@ -153,6 +153,7 @@ "vitest": "4.1.8" }, "overrides": { + "ast-v8-to-istanbul": "1.0.4", "dompurify": "3.4.10", "next": "16.2.6", "postcss": "8.5.12",