diff --git a/convex/crons.ts b/convex/crons.ts index 56b41067..cbdc6818 100644 --- a/convex/crons.ts +++ b/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 }, diff --git a/convex/lib/observabilityEvents.ts b/convex/lib/observabilityEvents.ts index aa4602b1..59d4b66a 100644 --- a/convex/lib/observabilityEvents.ts +++ b/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]; diff --git a/convex/securityScan.test.ts b/convex/securityScan.test.ts index ea0aced3..19843070 100644 --- a/convex/securityScan.test.ts +++ b/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, + { + snapshotAt: number; + queueDepth: number; + queueDepthIsEstimate: boolean; + readyQueueDepth: number; + readyQueueDepthIsEstimate: boolean; + oldestReadyJobAgeSeconds: number; + oldestReadyJobNextRunAt: number | null; + } + > +)._handler; + +const logCodexScanQueueHealthInternalHandler = ( + logCodexScanQueueHealthInternal as unknown as WrappedHandler< + Record, + { + 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 { }; } +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(); + 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", diff --git a/convex/securityScan.ts b/convex/securityScan.ts index 83da3135..04f5c9ea 100644 --- a/convex/securityScan.ts +++ b/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 => { + const snapshot = await runQueryRef( + 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;