feat: add ClawScan queue backlog telemetry (#3025)

convex/crons.ts

@@ -162,6 +162,13 @@ if (process.env.CLAWHUB_DISABLE_CRONS !== "1") {
     { batchSize: 10 },
   );
 
+  crons.interval(
+    "codex-scan-queue-health",
+    { minutes: 5 },
+    internal.securityScan.logCodexScanQueueHealthInternal,
+    {},
+  );
+
   crons.interval(
     "download-metric-dedupe-prune",
     { hours: 24 },

convex/lib/observabilityEvents.ts

@@ -3,6 +3,7 @@ export const Events = {
   GitHubSkillSourceSyncCompleted: "github_skill_source_sync.completed",
   GitHubSkillSourceSyncSourceFailed: "github_skill_source_sync.source_failed",
   GitHubSkillSourceSyncFailed: "github_skill_source_sync.failed",
+  SecurityScanQueueSnapshot: "security_scan_queue.snapshot",
 } as const;
 
 export type EventName = (typeof Events)[keyof typeof Events];

convex/securityScan.test.ts

@@ -12,10 +12,12 @@ import {
   failCodexScanJob,
   failJobInternal,
   finalizeGitHubSkillScanRequestInternal,
+  getCodexScanQueueHealthInternal,
   getJobTargetInternal,
   getBulkSkillRescanBatchStatusForAdminInternal,
   getSkillScanRequestForUserInternal,
   getStoredScanReportForUserInternal,
+  logCodexScanQueueHealthInternal,
   prepareGitHubSkillScanRequestInternal,
   pruneExpiredSkillScanRequestsInternal,
   recordGitHubSkillScanResultInternal,
@@ -63,6 +65,36 @@ const failJobInternalHandler = (
   >
 )._handler;
 
+const getCodexScanQueueHealthInternalHandler = (
+  getCodexScanQueueHealthInternal as unknown as WrappedHandler<
+    Record<string, never>,
+    {
+      snapshotAt: number;
+      queueDepth: number;
+      queueDepthIsEstimate: boolean;
+      readyQueueDepth: number;
+      readyQueueDepthIsEstimate: boolean;
+      oldestReadyJobAgeSeconds: number;
+      oldestReadyJobNextRunAt: number | null;
+    }
+  >
+)._handler;
+
+const logCodexScanQueueHealthInternalHandler = (
+  logCodexScanQueueHealthInternal as unknown as WrappedHandler<
+    Record<string, never>,
+    {
+      snapshotAt: number;
+      queueDepth: number;
+      queueDepthIsEstimate: boolean;
+      readyQueueDepth: number;
+      readyQueueDepthIsEstimate: boolean;
+      oldestReadyJobAgeSeconds: number;
+      oldestReadyJobNextRunAt: number | null;
+    }
+  >
+)._handler;
+
 const recordSkillScanRequestFailedInternalHandler = (
   recordSkillScanRequestFailedInternal as unknown as WrappedHandler<
     { scanId: string; error: string; llmAnalysis?: { status: string; checkedAt: number } },
@@ -457,6 +489,52 @@ function makeScanJob(overrides: Partial<ScanJob> = {}): ScanJob {
   };
 }
 
+function makeQueueHealthCtx(jobs: ScanJob[]) {
+  const query = vi.fn((table: string) => {
+    expect(table).toBe("securityScanJobs");
+    return {
+      withIndex: vi.fn(
+        (
+          indexName: string,
+          buildRange: (q: { eq: (field: string, value: unknown) => unknown }) => unknown,
+        ) => {
+          expect(indexName).toBe("by_status_and_next_run_at");
+          const equals = new Map<string, unknown>();
+          const range = {
+            eq(field: string, value: unknown) {
+              equals.set(field, value);
+              return range;
+            },
+          };
+          buildRange(range);
+          const matched = [...jobs]
+            .filter((job) =>
+              Array.from(equals.entries()).every(([field, value]) => {
+                return job[field as keyof ScanJob] === value;
+              }),
+            )
+            .sort(
+              (a, b) =>
+                a.nextRunAt - b.nextRunAt ||
+                a._creationTime - b._creationTime ||
+                a._id.localeCompare(b._id),
+            );
+          return {
+            order: vi.fn((direction: string) => {
+              expect(direction).toBe("asc");
+              return {
+                take: vi.fn(async (limit: number) => matched.slice(0, limit)),
+              };
+            }),
+          };
+        },
+      ),
+    };
+  });
+
+  return { db: { query } };
+}
+
 function makeTarget(llmStatus?: string) {
   if (!llmStatus) return {};
   return {
@@ -995,10 +1073,96 @@ function makeStoredScanReportCtx(options: {
 
 describe("securityScan", () => {
   afterEach(() => {
+    vi.useRealTimers();
     vi.unstubAllEnvs();
     vi.mocked(getAuthUserId).mockReset();
   });
 
+  it("reports claimable queue depth and oldest overdue age", async () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(1_000_000);
+    const ctx = makeQueueHealthCtx([
+      makeScanJob({
+        _id: "securityScanJobs:oldest-ready",
+        nextRunAt: 100_000,
+      }),
+      makeScanJob({
+        _id: "securityScanJobs:ready",
+        nextRunAt: 900_000,
+      }),
+      makeScanJob({
+        _id: "securityScanJobs:future",
+        nextRunAt: 1_100_000,
+      }),
+      makeScanJob({
+        _id: "securityScanJobs:running",
+        status: "running",
+        nextRunAt: 1,
+      }),
+    ]);
+
+    const result = await getCodexScanQueueHealthInternalHandler(ctx, {});
+
+    expect(result).toEqual({
+      snapshotAt: 1_000_000,
+      queueDepth: 3,
+      queueDepthIsEstimate: false,
+      readyQueueDepth: 2,
+      readyQueueDepthIsEstimate: false,
+      oldestReadyJobAgeSeconds: 900,
+      oldestReadyJobNextRunAt: 100_000,
+    });
+  });
+
+  it("marks capped queue health counts as estimates", async () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(1_000_000);
+    const ctx = makeQueueHealthCtx(
+      Array.from({ length: 513 }, (_, index) =>
+        makeScanJob({
+          _id: `securityScanJobs:queued-${index}`,
+          _creationTime: index,
+          nextRunAt: index,
+        }),
+      ),
+    );
+
+    const result = await getCodexScanQueueHealthInternalHandler(ctx, {});
+
+    expect(result).toMatchObject({
+      queueDepth: 512,
+      queueDepthIsEstimate: true,
+      readyQueueDepth: 512,
+      readyQueueDepthIsEstimate: true,
+    });
+  });
+
+  it("logs the queue health snapshot as a structured observability event", async () => {
+    const snapshot = {
+      snapshotAt: 1_000_000,
+      queueDepth: 4,
+      queueDepthIsEstimate: false,
+      readyQueueDepth: 2,
+      readyQueueDepthIsEstimate: false,
+      oldestReadyJobAgeSeconds: 901,
+      oldestReadyJobNextRunAt: 99_000,
+    };
+    const runQuery = vi.fn(async () => snapshot);
+    const log = vi.spyOn(console, "log").mockImplementation(() => undefined);
+
+    const result = await logCodexScanQueueHealthInternalHandler({ runQuery }, {});
+
+    expect(result).toEqual(snapshot);
+    expect(runQuery).toHaveBeenCalledWith(expect.anything(), {});
+    expect(log).toHaveBeenCalledWith(
+      JSON.stringify({
+        event: "security_scan_queue.snapshot",
+        ...snapshot,
+      }),
+    );
+    log.mockRestore();
+  });
+
   it("does not enqueue a duplicate publish scan after the backup delay if the first scan already finished", async () => {
     const existingJob = makeScanJob({
       _id: "securityScanJobs:fast-publish",

convex/securityScan.ts

@@ -2,9 +2,10 @@ import { ConvexError, v } from "convex/values";
 import { internal } from "./_generated/api";
 import type { Doc, Id } from "./_generated/dataModel";
 import type { MutationCtx, QueryCtx } from "./_generated/server";
-import { action, internalMutation, internalQuery, mutation } from "./functions";
+import { action, internalAction, internalMutation, internalQuery, mutation } from "./functions";
 import { applyGitHubSkillVerificationResultHandler } from "./githubSkillSync";
 import { assertAdmin, assertModerator, requireUser } from "./lib/access";
+import { Events, logEvent } from "./lib/observabilityEvents";
 import { normalizePackageName } from "./lib/packageRegistry";
 import { normalizePackageScanStatus } from "./lib/packageSecurity";
 import { assertCanManageOwnedResource } from "./lib/publishers";
@@ -44,6 +45,7 @@ const MAX_STORED_SKILLSPECTOR_SHORT_TEXT_CHARS = 512;
 const DEFAULT_SKILL_SCAN_REQUEST_RETENTION_MS = 7 * 24 * 60 * 60 * 1000;
 const MAX_SKILL_SCAN_QUEUE_POSITION_READS = 250;
 const MAX_SKILL_SCAN_RUNNING_COUNT_READS = 512;
+const MAX_SECURITY_SCAN_QUEUE_HEALTH_READS = 512;
 const GITHUB_SKILL_SCAN_ACTION_LEASE_MS = 15 * 60 * 1000;
 const SKILL_SCAN_ASYNC_NOTE = "Scans are asynchronous and may take time to complete.";
 
@@ -136,6 +138,16 @@ const jobSourceValidator = v.union(
 
 type SecurityScanJobSource = "publish" | "vt-update" | "backfill" | "bulk-rescan" | "manual";
 
+type CodexScanQueueHealth = {
+  snapshotAt: number;
+  queueDepth: number;
+  queueDepthIsEstimate: boolean;
+  readyQueueDepth: number;
+  readyQueueDepthIsEstimate: boolean;
+  oldestReadyJobAgeSeconds: number;
+  oldestReadyJobNextRunAt: number | null;
+};
+
 const CLAIM_SOURCE_ORDER: SecurityScanJobSource[] = [
   "backfill",
   "publish",
@@ -293,6 +305,7 @@ const internalRefs = internal as unknown as {
     enqueuePackageReleaseScanInternal: unknown;
     enqueueSkillVersionScanInternal: unknown;
     failJobInternal: unknown;
+    getCodexScanQueueHealthInternal: unknown;
     getSkillScanRequestForUserInternal: unknown;
     getJobTargetInternal: unknown;
     recordGitHubSkillScanResultInternal: unknown;
@@ -1048,6 +1061,49 @@ async function countSecurityScanJobs(
   };
 }
 
+export const getCodexScanQueueHealthInternal = internalQuery({
+  args: {},
+  handler: async (ctx) => {
+    const snapshotAt = Date.now();
+    const queuedJobs = await ctx.db
+      .query("securityScanJobs")
+      .withIndex("by_status_and_next_run_at", (q) => q.eq("status", "queued"))
+      .order("asc")
+      .take(MAX_SECURITY_SCAN_QUEUE_HEALTH_READS + 1);
+    const sampledJobs = queuedJobs.slice(0, MAX_SECURITY_SCAN_QUEUE_HEALTH_READS);
+    const firstFutureJobIndex = sampledJobs.findIndex((job) => job.nextRunAt > snapshotAt);
+    const readyQueueDepth = firstFutureJobIndex === -1 ? sampledJobs.length : firstFutureJobIndex;
+    const queueDepthIsEstimate = queuedJobs.length > MAX_SECURITY_SCAN_QUEUE_HEALTH_READS;
+    const oldestReadyJob = readyQueueDepth > 0 ? sampledJobs[0] : null;
+
+    return {
+      snapshotAt,
+      queueDepth: sampledJobs.length,
+      queueDepthIsEstimate,
+      readyQueueDepth,
+      readyQueueDepthIsEstimate:
+        queueDepthIsEstimate && readyQueueDepth === MAX_SECURITY_SCAN_QUEUE_HEALTH_READS,
+      oldestReadyJobAgeSeconds: oldestReadyJob
+        ? Math.max(0, Math.floor((snapshotAt - oldestReadyJob.nextRunAt) / 1000))
+        : 0,
+      oldestReadyJobNextRunAt: oldestReadyJob?.nextRunAt ?? null,
+    };
+  },
+});
+
+export const logCodexScanQueueHealthInternal = internalAction({
+  args: {},
+  handler: async (ctx): Promise<CodexScanQueueHealth> => {
+    const snapshot = await runQueryRef<CodexScanQueueHealth>(
+      ctx,
+      internalRefs.securityScan.getCodexScanQueueHealthInternal,
+      {},
+    );
+    logEvent(Events.SecurityScanQueueSnapshot, snapshot);
+    return snapshot;
+  },
+});
+
 function compareQueuedScanClaimOrder(a: Doc<"securityScanJobs">, b: Doc<"securityScanJobs">) {
   if (a.nextRunAt !== b.nextRunAt) return a.nextRunAt - b.nextRunAt;
   if (a._creationTime !== b._creationTime) return a._creationTime - b._creationTime;