fix: floor skill stat drain batch size

Floor legacy/small scheduled skill stat doc-sync action batch sizes to the production drain size so continuations cannot crawl indefinitely.

convex/skillStatEvents.test.ts

@@ -19,7 +19,8 @@ vi.mock("./_generated/api", () => ({
   },
 }));
 
-const { processSkillStatEventBatchInternal } = await import("./skillStatEvents");
+const { processSkillStatEventBatchInternal, processSkillStatEventsInternal } =
+  await import("./skillStatEvents");
 
 const processSkillStatEventBatchInternalHandler = (
   processSkillStatEventBatchInternal as unknown as {
@@ -30,6 +31,15 @@ const processSkillStatEventBatchInternalHandler = (
   }
 )._handler;
 
+const processSkillStatEventsInternalHandler = (
+  processSkillStatEventsInternal as unknown as {
+    _handler: (
+      ctx: unknown,
+      args: { batchSize?: number; maxBatches?: number },
+    ) => Promise<{ processed: number; scheduledContinuation: boolean }>;
+  }
+)._handler;
+
 // Test the aggregateEvents function by importing and testing the module logic
 // Since aggregateEvents is not exported, we test the behavior indirectly through
 // the event processing contract
@@ -112,6 +122,40 @@ describe("skill stat events - comment delta handling", () => {
     );
   });
 
+  it("floors action drain batch size so stale small continuations do not crawl", async () => {
+    const runMutation = vi.fn(async (_ref: unknown, args: Record<string, unknown>) => {
+      if ("leaseMs" in args) {
+        return {
+          acquired: true,
+          leaseOwner: "test-lease",
+          leaseExpiresAt: Date.now() + 60_000,
+          now: Date.now(),
+        };
+      }
+      if ("leaseOwner" in args && "batchSize" in args) {
+        return { processed: 100, skillsUpdated: 1, hasMore: true };
+      }
+      if ("processed" in args) {
+        return { released: true };
+      }
+      throw new Error(`unexpected mutation args ${JSON.stringify(args)}`);
+    });
+    const scheduler = { runAfter: vi.fn() };
+
+    await expect(
+      processSkillStatEventsInternalHandler(
+        { runMutation, scheduler },
+        { batchSize: 10, maxBatches: 1 },
+      ),
+    ).resolves.toMatchObject({
+      processed: 100,
+      scheduledContinuation: true,
+    });
+
+    expect(runMutation.mock.calls[1]?.[1]).toMatchObject({ batchSize: 100 });
+    expect(scheduler.runAfter.mock.calls[0]?.[2]).toMatchObject({ batchSize: 100 });
+  });
+
   it("aggregates comment and uncomment events into net deltas", () => {
     // Simulate the aggregation logic from processSkillStatEventsAction
     type EventKind =

convex/skillStatEvents.ts

@@ -233,6 +233,14 @@ function normalizeDocSyncBatchSize(batchSize: number | undefined) {
   return clampInt(batchSize ?? DEFAULT_DOC_SYNC_BATCH_SIZE, 1, MAX_DOC_SYNC_BATCH_SIZE);
 }
 
+function normalizeDocSyncDrainBatchSize(batchSize: number | undefined) {
+  return clampInt(
+    batchSize ?? DEFAULT_DOC_SYNC_BATCH_SIZE,
+    DEFAULT_DOC_SYNC_BATCH_SIZE,
+    MAX_DOC_SYNC_BATCH_SIZE,
+  );
+}
+
 function normalizeDocSyncMaxBatches(maxBatches: number | undefined) {
   return clampInt(maxBatches ?? DEFAULT_DOC_SYNC_MAX_BATCHES, 1, MAX_DOC_SYNC_MAX_BATCHES);
 }
@@ -447,7 +455,9 @@ export const processSkillStatEventsInternal: ReturnType<typeof internalAction> =
     maxBatches: v.optional(v.number()),
   },
   handler: async (ctx, args): Promise<SkillStatDocSyncActionResult> => {
-    const batchSize = normalizeDocSyncBatchSize(args.batchSize);
+    // Older scheduled continuations may carry a tiny batch size. Floor action
+    // drains to the production batch size so stale jobs cannot crawl forever.
+    const batchSize = normalizeDocSyncDrainBatchSize(args.batchSize);
     const maxBatches = normalizeDocSyncMaxBatches(args.maxBatches);
     const claim: ClaimSkillStatDocSyncLeaseResult = await ctx.runMutation(
       internal.skillStatEvents.claimSkillStatDocSyncLeaseInternal,
@@ -530,7 +540,7 @@ export const kickSkillStatDocSyncInternal = internalMutation({
     maxBatches: v.optional(v.number()),
   },
   handler: async (ctx, args) => {
-    const batchSize = normalizeDocSyncBatchSize(args.batchSize);
+    const batchSize = normalizeDocSyncDrainBatchSize(args.batchSize);
     const maxBatches = normalizeDocSyncMaxBatches(args.maxBatches);
     await ctx.scheduler.runAfter(0, internal.skillStatEvents.processSkillStatEventsInternal, {
       batchSize,