fix(convex): reduce download and token write contention

convex/downloads.test.ts

@@ -62,7 +62,7 @@ describe("downloads helpers", () => {
     expect(__test.getDownloadIdentityValue(request, null)).toBeNull();
   });
 
-  it("records zip downloads through the internal mutation path", async () => {
+  it("schedules zip download stats outside the response path", async () => {
     class MockResponse {
       status: number;
       headers: Headers;
@@ -103,12 +103,14 @@ describe("downloads helpers", () => {
       if (isRateLimitArgs(args)) return okRate();
       return { mutation, args };
     });
+    const runAfter = vi.fn();
     const storageGet = vi.fn().mockResolvedValue(new Blob(["hello"], { type: "text/markdown" }));
 
     const response = await downloadZipHandler(
       {
         runQuery,
         runMutation,
+        scheduler: { runAfter },
         storage: { get: storageGet },
       } as unknown as ActionCtx,
       new Request("https://example.com/api/v1/download?slug=demo", {
@@ -120,7 +122,7 @@ describe("downloads helpers", () => {
     expect(response.headers.get("Content-Type")).toBe("application/zip");
     expect(storageGet).toHaveBeenCalledWith("_storage:1");
 
-    const recordCalls = runMutation.mock.calls.filter(([, args]) => {
+    const recordCalls = runAfter.mock.calls.filter(([, , args]) => {
       if (!args || typeof args !== "object") return false;
       const value = args as Record<string, unknown>;
       return (
@@ -130,7 +132,10 @@ describe("downloads helpers", () => {
       );
     });
     expect(recordCalls).toHaveLength(1);
-    expect(recordCalls[0]?.[1]).toEqual({
+    expect(recordCalls[0]?.[0]).toEqual(expect.any(Number));
+    expect(recordCalls[0]?.[0]).toBeGreaterThanOrEqual(0);
+    expect(recordCalls[0]?.[0]).toBeLessThan(60_000);
+    expect(recordCalls[0]?.[2]).toEqual({
       skillId: "skills:1",
       identityHash: expect.any(String),
       hourStart: expect.any(Number),

convex/downloads.ts

@@ -12,6 +12,7 @@ const HOUR_MS = 3_600_000;
 const DEDUPE_RETENTION_MS = 7 * 24 * HOUR_MS;
 const PRUNE_BATCH_SIZE = 200;
 const PRUNE_MAX_BATCHES = 50;
+const DOWNLOAD_STAT_JITTER_MS = 60_000;
 
 export async function downloadZipHandler(
   ctx: Parameters<Parameters<typeof httpAction>[0]>[0],
@@ -124,11 +125,15 @@ export async function downloadZipHandler(
     const userId = await getOptionalApiTokenUserId(ctx, request);
     const identity = getDownloadIdentityValue(request, userId ? String(userId) : null);
     if (identity) {
-      await ctx.runMutation(internal.downloads.recordDownloadInternal, {
-        skillId: skill._id,
-        identityHash: await hashToken(identity),
-        hourStart: getHourStart(Date.now()),
-      });
+      await ctx.scheduler.runAfter(
+        Math.floor(Math.random() * DOWNLOAD_STAT_JITTER_MS),
+        internal.downloads.recordDownloadInternal,
+        {
+          skillId: skill._id,
+          identityHash: await hashToken(identity),
+          hourStart: getHourStart(Date.now()),
+        },
+      );
     }
   } catch {
     // Best-effort metric path; do not fail downloads.
@@ -165,7 +170,7 @@ export const recordDownloadInternal = internalMutation({
           .eq("identityHash", args.identityHash)
           .eq("hourStart", args.hourStart),
       )
-      .unique();
+      .first();
     if (existing) return;
 
     await ctx.db.insert("downloadDedupes", {

convex/lib/apiTokenAuth.ts

@@ -54,10 +54,14 @@ export async function requireApiTokenUser(
   )) as Doc<"users"> | null;
   if (!user || user.deletedAt || user.deactivatedAt) throw new ConvexError("Unauthorized");
 
-  await ctx.runMutation(
-    internalRefs.tokens.touchInternal as never,
-    { tokenId: apiToken._id } as never,
-  );
+  try {
+    await ctx.runMutation(
+      internalRefs.tokens.touchInternal as never,
+      { tokenId: apiToken._id } as never,
+    );
+  } catch {
+    // Best-effort metadata; auth succeeded and should not fail on write contention.
+  }
   return { user, userId: user._id };
 }
 
@@ -105,12 +109,16 @@ export async function requirePackagePublishAuth(
     } as never,
   )) as PackagePublishTokenDoc | null;
   if (publishToken && !publishToken.revokedAt && publishToken.expiresAt > Date.now()) {
-    await ctx.runMutation(
-      internalRefs.packagePublishTokens.touchInternal as never,
-      {
-        tokenId: publishToken._id,
-      } as never,
-    );
+    try {
+      await ctx.runMutation(
+        internalRefs.packagePublishTokens.touchInternal as never,
+        {
+          tokenId: publishToken._id,
+        } as never,
+      );
+    } catch {
+      // Best-effort metadata; publish auth should not fail on touch contention.
+    }
     return { kind: "github-actions", publishToken };
   }
 

convex/packagePublishTokens.ts

@@ -1,6 +1,8 @@
 import { v } from "convex/values";
 import { internalMutation, internalQuery } from "./functions";
 
+const TOKEN_TOUCH_MIN_INTERVAL_MS = 15 * 60_000;
+
 export const createInternal = internalMutation({
   args: {
     packageId: v.id("packages"),
@@ -54,9 +56,11 @@ export const getByIdInternal = internalQuery({
 export const touchInternal = internalMutation({
   args: { tokenId: v.id("packagePublishTokens") },
   handler: async (ctx, args) => {
+    const now = Date.now();
     const token = await ctx.db.get(args.tokenId);
-    if (!token || token.revokedAt || token.expiresAt <= Date.now()) return;
-    await ctx.db.patch(token._id, { lastUsedAt: Date.now() });
+    if (!token || token.revokedAt || token.expiresAt <= now) return;
+    if (token.lastUsedAt && now - token.lastUsedAt < TOKEN_TOUCH_MIN_INTERVAL_MS) return;
+    await ctx.db.patch(token._id, { lastUsedAt: now });
   },
 });
 

convex/tokens.touch.test.ts

@@ -0,0 +1,76 @@
+/* @vitest-environment node */
+
+import { describe, expect, it, vi } from "vitest";
+import { touchInternal as touchPackagePublishTokenInternal } from "./packagePublishTokens";
+import { touchInternal as touchApiTokenInternal } from "./tokens";
+
+type WrappedHandler<TArgs> = {
+  _handler: (ctx: unknown, args: TArgs) => Promise<void>;
+};
+
+const touchApiTokenHandler = (
+  touchApiTokenInternal as unknown as WrappedHandler<{ tokenId: string }>
+)._handler;
+const touchPackagePublishTokenHandler = (
+  touchPackagePublishTokenInternal as unknown as WrappedHandler<{ tokenId: string }>
+)._handler;
+
+function makeCtx(token: Record<string, unknown> | null) {
+  return {
+    db: {
+      get: vi.fn(async () => token),
+      insert: vi.fn(),
+      normalizeId: vi.fn(),
+      patch: vi.fn(),
+      query: vi.fn(),
+      replace: vi.fn(),
+      delete: vi.fn(),
+      system: {
+        get: vi.fn(),
+        query: vi.fn(),
+      },
+    },
+  };
+}
+
+describe("token touch throttling", () => {
+  it("skips api token touches inside the freshness window", async () => {
+    vi.spyOn(Date, "now").mockReturnValue(1_000_000);
+    const ctx = makeCtx({
+      _id: "apiTokens:one",
+      revokedAt: undefined,
+      lastUsedAt: 500_000,
+    });
+
+    await touchApiTokenHandler(ctx, { tokenId: "apiTokens:one" });
+
+    expect(ctx.db.patch).not.toHaveBeenCalled();
+  });
+
+  it("patches stale api token touches", async () => {
+    vi.spyOn(Date, "now").mockReturnValue(1_000_000);
+    const ctx = makeCtx({
+      _id: "apiTokens:one",
+      revokedAt: undefined,
+      lastUsedAt: 1,
+    });
+
+    await touchApiTokenHandler(ctx, { tokenId: "apiTokens:one" });
+
+    expect(ctx.db.patch).toHaveBeenCalledWith("apiTokens:one", { lastUsedAt: 1_000_000 });
+  });
+
+  it("skips package publish token touches inside the freshness window", async () => {
+    vi.spyOn(Date, "now").mockReturnValue(1_000_000);
+    const ctx = makeCtx({
+      _id: "packagePublishTokens:one",
+      revokedAt: undefined,
+      expiresAt: 2_000_000,
+      lastUsedAt: 500_000,
+    });
+
+    await touchPackagePublishTokenHandler(ctx, { tokenId: "packagePublishTokens:one" });
+
+    expect(ctx.db.patch).not.toHaveBeenCalled();
+  });
+});

convex/tokens.ts

@@ -4,6 +4,8 @@ import { internalMutation, internalQuery, mutation, query } from "./functions";
 import { requireUser } from "./lib/access";
 import { generateToken, hashToken } from "./lib/tokens";
 
+const TOKEN_TOUCH_MIN_INTERVAL_MS = 15 * 60_000;
+
 export const listMine = query({
   args: {},
   handler: async (ctx) => {
@@ -72,9 +74,11 @@ export const getByHashInternal = internalQuery({
 export const touchInternal = internalMutation({
   args: { tokenId: v.id("apiTokens") },
   handler: async (ctx, args) => {
+    const now = Date.now();
     const token = await ctx.db.get(args.tokenId);
     if (!token || token.revokedAt) return;
-    await ctx.db.patch(token._id, { lastUsedAt: Date.now() });
+    if (token.lastUsedAt && now - token.lastUsedAt < TOKEN_TOUCH_MIN_INTERVAL_MS) return;
+    await ctx.db.patch(token._id, { lastUsedAt: now });
   },
 });