fix(convex): reduce hot stat write contention

convex/crons.ts

@@ -33,6 +33,13 @@ crons.interval(
   {},
 );
 
+crons.interval(
+  "package-stat-events",
+  { minutes: 15 },
+  internal.packages.processPackageStatEventsInternal,
+  { batchSize: 500 },
+);
+
 // Syncs accumulated stat deltas to skill documents every 6 hours.
 // Runs infrequently to avoid thundering-herd reactive query invalidation.
 // Uses processedAt field to track progress (independent of the action cursor).
@@ -40,7 +47,7 @@ crons.interval(
   "skill-doc-stat-sync",
   { hours: 6 },
   internal.skillStatEvents.processSkillStatEventsInternal,
-  { batchSize: 500 },
+  { batchSize: 100 },
 );
 
 crons.interval(

convex/lib/httpRateLimit.ts

@@ -4,6 +4,7 @@ import { getOptionalApiTokenUserId } from "./apiTokenAuth";
 import { corsHeaders, mergeHeaders } from "./httpHeaders";
 
 const RATE_LIMIT_WINDOW_MS = 60_000;
+const RATE_LIMIT_SHARDS = 16;
 export const RATE_LIMITS = {
   read: { ip: 600, key: 2400 },
   write: { ip: 45, key: 180 },
@@ -147,6 +148,7 @@ async function checkRateLimit(
       key,
       limit,
       windowMs: RATE_LIMIT_WINDOW_MS,
+      shard: Math.floor(Math.random() * RATE_LIMIT_SHARDS),
     })) as { allowed: boolean; remaining: number };
   } catch (error) {
     if (isRateLimitWriteConflict(error)) {
@@ -162,7 +164,7 @@ async function checkRateLimit(
 
   return {
     allowed: result.allowed,
-    remaining: result.remaining,
+    remaining: Math.max(0, status.remaining - 1),
     limit: status.limit,
     resetAt: status.resetAt,
   };

convex/packages.stats.test.ts

@@ -0,0 +1,109 @@
+/* @vitest-environment node */
+
+import { describe, expect, it, vi } from "vitest";
+import { processPackageStatEventsInternal, recordPackageDownloadInternal } from "./packages";
+
+type WrappedHandler<TArgs, TResult> = {
+  _handler: (ctx: unknown, args: TArgs) => Promise<TResult>;
+};
+
+const recordDownloadHandler = (
+  recordPackageDownloadInternal as unknown as WrappedHandler<{ packageId: string }, void>
+)._handler;
+
+const processStatsHandler = (
+  processPackageStatEventsInternal as unknown as WrappedHandler<
+    { batchSize?: number },
+    { processed: number; packagesUpdated: number }
+  >
+)._handler;
+
+describe("package stat events", () => {
+  it("records downloads as append-only events", async () => {
+    const insert = vi.fn();
+
+    await recordDownloadHandler(
+      {
+        db: {
+          query: vi.fn(),
+          get: vi.fn(),
+          normalizeId: vi.fn(),
+          insert,
+          patch: vi.fn(),
+          replace: vi.fn(),
+          delete: vi.fn(),
+          system: {
+            get: vi.fn(),
+            query: vi.fn(),
+          },
+        },
+      },
+      {
+        packageId: "packages:one",
+      },
+    );
+
+    expect(insert).toHaveBeenCalledWith(
+      "packageStatEvents",
+      expect.objectContaining({
+        packageId: "packages:one",
+        kind: "download",
+        processedAt: undefined,
+      }),
+    );
+  });
+
+  it("aggregates queued downloads before patching package stats", async () => {
+    const events = [
+      { _id: "packageStatEvents:1", packageId: "packages:one" },
+      { _id: "packageStatEvents:2", packageId: "packages:one" },
+      { _id: "packageStatEvents:3", packageId: "packages:two" },
+    ];
+    const patch = vi.fn();
+    const ctx = {
+      db: {
+        query: vi.fn(() => ({
+          withIndex: vi.fn(() => ({
+            take: vi.fn(async () => events),
+          })),
+        })),
+        get: vi.fn(async (id: string) => ({
+          _id: id,
+          stats: { downloads: 10, installs: 1, stars: 2, versions: 3 },
+        })),
+        normalizeId: vi.fn(),
+        insert: vi.fn(),
+        patch,
+        replace: vi.fn(),
+        delete: vi.fn(),
+        system: {
+          get: vi.fn(),
+          query: vi.fn(),
+        },
+      },
+      scheduler: {
+        runAfter: vi.fn(),
+      },
+    };
+
+    const result = await processStatsHandler(ctx, { batchSize: 10 });
+
+    expect(result).toEqual({ processed: 3, packagesUpdated: 2 });
+    expect(patch).toHaveBeenCalledWith(
+      "packages:one",
+      expect.objectContaining({
+        stats: expect.objectContaining({ downloads: 12 }),
+      }),
+    );
+    expect(patch).toHaveBeenCalledWith(
+      "packages:two",
+      expect.objectContaining({
+        stats: expect.objectContaining({ downloads: 11 }),
+      }),
+    );
+    expect(patch).toHaveBeenCalledWith(
+      "packageStatEvents:1",
+      expect.objectContaining({ processedAt: expect.any(Number) }),
+    );
+  });
+});

convex/packages.ts

@@ -1911,19 +1911,61 @@ export const getPackageByNameInternal = internalQuery({
 export const recordPackageDownloadInternal = internalMutation({
   args: { packageId: v.id("packages") },
   handler: async (ctx, args) => {
-    const pkg = await ctx.db.get(args.packageId);
-    if (!pkg) return;
-    await ctx.db.patch(pkg._id, {
-      stats: {
-        downloads: (pkg.stats?.downloads ?? 0) + 1,
-        installs: pkg.stats?.installs ?? 0,
-        stars: pkg.stats?.stars ?? 0,
-        versions: pkg.stats?.versions ?? 0,
-      },
+    await ctx.db.insert("packageStatEvents", {
+      packageId: args.packageId,
+      kind: "download",
+      occurredAt: Date.now(),
+      processedAt: undefined,
     });
   },
 });
 
+export const processPackageStatEventsInternal = internalMutation({
+  args: { batchSize: v.optional(v.number()) },
+  handler: async (ctx, args) => {
+    const batchSize = Math.max(1, Math.min(args.batchSize ?? 500, 1_000));
+    const now = Date.now();
+    const events = await ctx.db
+      .query("packageStatEvents")
+      .withIndex("by_unprocessed", (q) => q.eq("processedAt", undefined))
+      .take(batchSize);
+
+    if (events.length === 0) return { processed: 0, packagesUpdated: 0 };
+
+    const downloadsByPackage = new Map<Id<"packages">, number>();
+    for (const event of events) {
+      downloadsByPackage.set(event.packageId, (downloadsByPackage.get(event.packageId) ?? 0) + 1);
+    }
+
+    let packagesUpdated = 0;
+    for (const [packageId, downloads] of downloadsByPackage) {
+      const pkg = await ctx.db.get(packageId);
+      if (!pkg) continue;
+      await ctx.db.patch(pkg._id, {
+        stats: {
+          downloads: (pkg.stats?.downloads ?? 0) + downloads,
+          installs: pkg.stats?.installs ?? 0,
+          stars: pkg.stats?.stars ?? 0,
+          versions: pkg.stats?.versions ?? 0,
+        },
+      });
+      packagesUpdated += 1;
+    }
+
+    for (const event of events) {
+      await ctx.db.patch(event._id, { processedAt: now });
+    }
+
+    if (events.length === batchSize) {
+      await ctx.scheduler.runAfter(0, internal.packages.processPackageStatEventsInternal, {
+        batchSize,
+      });
+    }
+
+    return { processed: events.length, packagesUpdated };
+  },
+});
+
 export const getTrustedPublisherByPackageIdInternal = internalQuery({
   args: { packageId: v.id("packages") },
   handler: async (ctx, args) => {

convex/rateLimits.test.ts

@@ -0,0 +1,91 @@
+/* @vitest-environment node */
+
+import { describe, expect, it, vi } from "vitest";
+import { consumeRateLimitInternal, getRateLimitStatusInternal } from "./rateLimits";
+
+type WrappedHandler<TArgs, TResult> = {
+  _handler: (ctx: unknown, args: TArgs) => Promise<TResult>;
+};
+
+const getStatusHandler = (
+  getRateLimitStatusInternal as unknown as WrappedHandler<
+    { key: string; limit: number; windowMs: number },
+    { allowed: boolean; remaining: number; limit: number; resetAt: number }
+  >
+)._handler;
+
+const consumeHandler = (
+  consumeRateLimitInternal as unknown as WrappedHandler<
+    { key: string; limit: number; windowMs: number; shard?: number },
+    { allowed: boolean; remaining: number }
+  >
+)._handler;
+
+describe("rate limit sharding", () => {
+  it("sums all shards when checking status", async () => {
+    const ctx = {
+      db: {
+        query: vi.fn(() => ({
+          withIndex: vi.fn(() => ({
+            collect: vi.fn(async () => [{ count: 3 }, { count: 4 }, { count: 5 }]),
+          })),
+        })),
+      },
+    };
+
+    const result = await getStatusHandler(ctx, {
+      key: "ip:test",
+      limit: 20,
+      windowMs: 60_000,
+    });
+
+    expect(result.allowed).toBe(true);
+    expect(result.remaining).toBe(8);
+  });
+
+  it("writes only the selected shard when consuming", async () => {
+    const insert = vi.fn();
+    const withIndex = vi.fn((_index, builder) => {
+      builder({
+        eq: vi.fn(() => ({
+          eq: vi.fn(() => ({
+            eq: vi.fn(),
+          })),
+        })),
+      });
+      return { unique: vi.fn(async () => null) };
+    });
+    const ctx = {
+      db: {
+        query: vi.fn(() => ({ withIndex })),
+        get: vi.fn(),
+        normalizeId: vi.fn(),
+        insert,
+        patch: vi.fn(),
+        replace: vi.fn(),
+        delete: vi.fn(),
+        system: {
+          get: vi.fn(),
+          query: vi.fn(),
+        },
+      },
+    };
+
+    await consumeHandler(ctx, {
+      key: "ip:test",
+      limit: 20,
+      windowMs: 60_000,
+      shard: 7,
+    });
+
+    expect(withIndex).toHaveBeenCalledWith("by_key_window_shard", expect.any(Function));
+    expect(insert).toHaveBeenCalledWith(
+      "rateLimits",
+      expect.objectContaining({
+        key: "ip:test",
+        shard: 7,
+        count: 1,
+      }),
+    );
+  });
+});

convex/rateLimits.ts

@@ -19,12 +19,12 @@ export const getRateLimitStatusInternal = internalQuery({
       return { allowed: false, remaining: 0, limit: args.limit, resetAt };
     }
 
-    const existing = await ctx.db
+    const rows = await ctx.db
       .query("rateLimits")
       .withIndex("by_key_window", (q) => q.eq("key", args.key).eq("windowStart", windowStart))
-      .unique();
+      .collect();
 
-    const count = existing?.count ?? 0;
+    const count = rows.reduce((sum, row) => sum + row.count, 0);
     const allowed = count < args.limit;
     return {
       allowed,
@@ -45,26 +45,25 @@ export const consumeRateLimitInternal = internalMutation({
     key: v.string(),
     limit: v.number(),
     windowMs: v.number(),
+    shard: v.optional(v.number()),
   },
   handler: async (ctx, args) => {
     const now = Date.now();
     const windowStart = Math.floor(now / args.windowMs) * args.windowMs;
+    const shard = Math.max(0, Math.floor(args.shard ?? 0));
 
     const existing = await ctx.db
       .query("rateLimits")
-      .withIndex("by_key_window", (q) => q.eq("key", args.key).eq("windowStart", windowStart))
+      .withIndex("by_key_window_shard", (q) =>
+        q.eq("key", args.key).eq("windowStart", windowStart).eq("shard", shard),
+      )
       .unique();
 
-    // Double-check: another request may have consumed the last token
-    // between our query and this mutation
-    if (existing && existing.count >= args.limit) {
-      return { allowed: false, remaining: 0 };
-    }
-
     if (!existing) {
       await ctx.db.insert("rateLimits", {
         key: args.key,
         windowStart,
+        shard,
         count: 1,
         limit: args.limit,
         updatedAt: now,
@@ -79,7 +78,7 @@ export const consumeRateLimitInternal = internalMutation({
     });
     return {
       allowed: true,
-      remaining: Math.max(0, args.limit - existing.count - 1),
+      remaining: Math.max(0, args.limit - 1),
     };
   },
 });

convex/schema.ts

@@ -914,6 +914,15 @@ const packageReleases = defineTable({
   .index("by_package_version", ["packageId", "version"])
   .index("by_sha256hash", ["sha256hash"]);
 
+const packageStatEvents = defineTable({
+  packageId: v.id("packages"),
+  kind: v.literal("download"),
+  occurredAt: v.number(),
+  processedAt: v.optional(v.number()),
+})
+  .index("by_unprocessed", ["processedAt"])
+  .index("by_package", ["packageId"]);
+
 const packageTrustedPublishers = defineTable({
   packageId: v.id("packages"),
   provider: v.literal("github-actions"),
@@ -1453,11 +1462,13 @@ const apiTokens = defineTable({
 const rateLimits = defineTable({
   key: v.string(),
   windowStart: v.number(),
+  shard: v.optional(v.number()),
   count: v.number(),
   limit: v.number(),
   updatedAt: v.number(),
 })
   .index("by_key_window", ["key", "windowStart"])
+  .index("by_key_window_shard", ["key", "windowStart", "shard"])
   .index("by_key", ["key"]);
 
 const downloadDedupes = defineTable({
@@ -1571,6 +1582,7 @@ export default defineSchema({
   skillSlugAliases,
   packages,
   packageReleases,
+  packageStatEvents,
   packageTrustedPublishers,
   packagePublishTokens,
   packageBadges,