From b0984d33c0792394ad62abfcf9093e619c5a8cb6 Mon Sep 17 00:00:00 2001 From: Patrick Erichsen Date: Mon, 20 Jul 2026 12:22:56 -0700 Subject: [PATCH] feat(observability): log prepublication queue health (#3192) --- convex/_generated/api.d.ts | 2 + convex/crons.test.ts | 16 +++ convex/crons.ts | 7 ++ convex/lib/observabilityEvents.ts | 1 + convex/prepublicationObservability.test.ts | 119 +++++++++++++++++++++ convex/prepublicationObservability.ts | 87 +++++++++++++++ 6 files changed, 232 insertions(+) create mode 100644 convex/prepublicationObservability.test.ts create mode 100644 convex/prepublicationObservability.ts diff --git a/convex/_generated/api.d.ts b/convex/_generated/api.d.ts index 5ac5f0df..096752ee 100644 --- a/convex/_generated/api.d.ts +++ b/convex/_generated/api.d.ts @@ -143,6 +143,7 @@ import type * as packageInspectorNode from "../packageInspectorNode.js"; import type * as packageLeaderboards from "../packageLeaderboards.js"; import type * as packagePublishTokens from "../packagePublishTokens.js"; import type * as packages from "../packages.js"; +import type * as prepublicationObservability from "../prepublicationObservability.js"; import type * as promotions from "../promotions.js"; import type * as promotionsFeed from "../promotionsFeed.js"; import type * as publishAttempts from "../publishAttempts.js"; @@ -312,6 +313,7 @@ declare const fullApi: ApiFromModules<{ packageLeaderboards: typeof packageLeaderboards; packagePublishTokens: typeof packagePublishTokens; packages: typeof packages; + prepublicationObservability: typeof prepublicationObservability; promotions: typeof promotions; promotionsFeed: typeof promotionsFeed; publishAttempts: typeof publishAttempts; diff --git a/convex/crons.test.ts b/convex/crons.test.ts index b0c2a36b..5fcec64c 100644 --- a/convex/crons.test.ts +++ b/convex/crons.test.ts @@ -17,6 +17,7 @@ const mocks = vi.hoisted(() => { const authRefreshTokensPruneRef = Symbol("auth-refresh-tokens-prune"); const publisherInvitesPruneRef = Symbol("publisher-invites-prune"); const promotionsFeedPublishRef = Symbol("promotions-feed-publish"); + const prepublicationQueueHealthRef = Symbol("prepublication-queue-health"); const securityScanExpiredLeaseRecoveryRef = Symbol("security-scan-expired-lease-recovery"); const securityScanDispatchWatchdogRef = Symbol("security-scan-dispatch-watchdog"); return { @@ -35,6 +36,7 @@ const mocks = vi.hoisted(() => { authRefreshTokensPruneRef, publisherInvitesPruneRef, promotionsFeedPublishRef, + prepublicationQueueHealthRef, securityScanExpiredLeaseRecoveryRef, securityScanDispatchWatchdogRef, }; @@ -80,6 +82,9 @@ vi.mock("./_generated/api", () => ({ promotionsFeed: { publishInternal: mocks.promotionsFeedPublishRef, }, + prepublicationObservability: { + logPrePublicationQueueHealthInternal: mocks.prepublicationQueueHealthRef, + }, vt: { pollPendingScans: Symbol("vt-pending-scans"), backfillActiveSkillsVTCache: Symbol("vt-cache-backfill"), @@ -181,6 +186,17 @@ describe("crons", () => { ); }); + it("logs pre-publication queue health every five minutes", async () => { + await import("./crons"); + + expect(mocks.interval).toHaveBeenCalledWith( + "prepublication-queue-health", + { minutes: 5 }, + mocks.prepublicationQueueHealthRef, + {}, + ); + }); + it("recovers expired security scan leases outside the claim hot path", async () => { await import("./crons"); diff --git a/convex/crons.ts b/convex/crons.ts index e92c5997..085adb9b 100644 --- a/convex/crons.ts +++ b/convex/crons.ts @@ -168,6 +168,13 @@ if (process.env.CLAWHUB_DISABLE_CRONS !== "1" && process.env.CLAWHUB_PREVIEW !== {}, ); + crons.interval( + "prepublication-queue-health", + { minutes: 5 }, + internal.prepublicationObservability.logPrePublicationQueueHealthInternal, + {}, + ); + crons.interval( "codex-scan-expired-lease-recovery", { minutes: 5 }, diff --git a/convex/lib/observabilityEvents.ts b/convex/lib/observabilityEvents.ts index 59d4b66a..c98fad1e 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", + PrePublicationQueueSnapshot: "prepublication_queue.snapshot", SecurityScanQueueSnapshot: "security_scan_queue.snapshot", } as const; diff --git a/convex/prepublicationObservability.test.ts b/convex/prepublicationObservability.test.ts new file mode 100644 index 00000000..3ce1d2c8 --- /dev/null +++ b/convex/prepublicationObservability.test.ts @@ -0,0 +1,119 @@ +import { describe, expect, it, vi } from "vitest"; +import { + getPrePublicationQueueHealthInternal, + logPrePublicationQueueHealthInternal, +} from "./prepublicationObservability"; + +const getQueueHealthHandler = ( + getPrePublicationQueueHealthInternal as unknown as { + _handler: (ctx: unknown, args: unknown) => Promise; + } +)._handler; +const logQueueHealthHandler = ( + logPrePublicationQueueHealthInternal as unknown as { + _handler: (ctx: unknown, args: unknown) => Promise; + } +)._handler; + +function makeQueueHealthCtx(attempts: Array>) { + return { + db: { + query: vi.fn((table: string) => { + expect(table).toBe("publishAttempts"); + return { + withIndex: vi.fn( + ( + indexName: string, + buildRange: (q: { eq: (field: string, value: unknown) => unknown }) => unknown, + ) => { + expect(indexName).toBe("by_status_and_created"); + const equals = new Map(); + const range = { + eq(field: string, value: unknown) { + equals.set(field, value); + return range; + }, + }; + buildRange(range); + const matched = attempts + .filter((attempt) => + Array.from(equals.entries()).every(([field, value]) => attempt[field] === value), + ) + .sort((a, b) => Number(a.createdAt) - Number(b.createdAt)); + return { + order: vi.fn((direction: string) => { + expect(direction).toBe("asc"); + return { + take: vi.fn(async (limit: number) => matched.slice(0, limit)), + }; + }), + }; + }, + ), + }; + }), + }, + }; +} + +describe("prepublication observability", () => { + it("reports timeout accumulation, active claims, and oldest ready age", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000_000); + const ctx = makeQueueHealthCtx([ + { + status: "pending_checks", + createdAt: 100_000, + checkClaimExpiresAt: 0, + checks: { clawscan: { status: "failed", summary: "clawscan timed out" } }, + }, + { + status: "pending_checks", + createdAt: 200_000, + checkClaimExpiresAt: 1_100_000, + checks: { clawscan: { status: "pending" } }, + }, + { + status: "finalized", + createdAt: 50_000, + checks: { clawscan: { status: "clean" } }, + }, + ]); + + await expect(getQueueHealthHandler(ctx, {})).resolves.toEqual({ + snapshotAt: 1_000_000, + pendingChecks: 2, + pendingChecksIsEstimate: false, + readyChecks: 1, + activeClaims: 1, + timeoutPending: 1, + scannerFailurePending: 1, + oldestPendingAgeSeconds: 900, + oldestReadyAgeSeconds: 900, + }); + }); + + it("logs a structured event for Axiom monitors", async () => { + const snapshot = { + snapshotAt: 1_000_000, + pendingChecks: 4, + pendingChecksIsEstimate: false, + readyChecks: 3, + activeClaims: 1, + timeoutPending: 2, + scannerFailurePending: 2, + oldestPendingAgeSeconds: 901, + oldestReadyAgeSeconds: 901, + }; + const runQuery = vi.fn(async () => snapshot); + const log = vi.spyOn(console, "log").mockImplementation(() => undefined); + + await expect(logQueueHealthHandler({ runQuery }, {})).resolves.toEqual(snapshot); + expect(log).toHaveBeenCalledWith( + JSON.stringify({ + event: "prepublication_queue.snapshot", + ...snapshot, + }), + ); + }); +}); diff --git a/convex/prepublicationObservability.ts b/convex/prepublicationObservability.ts new file mode 100644 index 00000000..3d504deb --- /dev/null +++ b/convex/prepublicationObservability.ts @@ -0,0 +1,87 @@ +import { internal } from "./_generated/api"; +import type { Doc } from "./_generated/dataModel"; +import { internalAction, internalQuery } from "./functions"; +import { Events, logEvent } from "./lib/observabilityEvents"; + +const MAX_PREPUBLICATION_QUEUE_HEALTH_READS = 512; + +type PrePublicationQueueHealth = { + snapshotAt: number; + pendingChecks: number; + pendingChecksIsEstimate: boolean; + readyChecks: number; + activeClaims: number; + timeoutPending: number; + scannerFailurePending: number; + oldestPendingAgeSeconds: number; + oldestReadyAgeSeconds: number; +}; + +const internalRefs = internal as unknown as { + prepublicationObservability: { + getPrePublicationQueueHealthInternal: unknown; + }; +}; + +async function runQueryRef( + ctx: { runQuery: (ref: never, args: never) => Promise }, + ref: unknown, + args: unknown, +): Promise { + return (await ctx.runQuery(ref as never, args as never)) as T; +} + +function isReady(attempt: Doc<"publishAttempts">, snapshotAt: number) { + return (attempt.checkClaimExpiresAt ?? 0) <= snapshotAt; +} + +function isClawScanTimeout(attempt: Doc<"publishAttempts">) { + return attempt.checks.clawscan.summary?.toLowerCase().includes("timed out") ?? false; +} + +export const getPrePublicationQueueHealthInternal = internalQuery({ + args: {}, + handler: async (ctx): Promise => { + const snapshotAt = Date.now(); + const pendingAttempts = await ctx.db + .query("publishAttempts") + .withIndex("by_status_and_created", (q) => q.eq("status", "pending_checks")) + .order("asc") + .take(MAX_PREPUBLICATION_QUEUE_HEALTH_READS + 1); + const sampledAttempts = pendingAttempts.slice(0, MAX_PREPUBLICATION_QUEUE_HEALTH_READS); + const readyAttempts = sampledAttempts.filter((attempt) => isReady(attempt, snapshotAt)); + const oldestPendingAttempt = sampledAttempts[0]; + const oldestReadyAttempt = readyAttempts[0]; + + return { + snapshotAt, + pendingChecks: sampledAttempts.length, + pendingChecksIsEstimate: pendingAttempts.length > MAX_PREPUBLICATION_QUEUE_HEALTH_READS, + readyChecks: readyAttempts.length, + activeClaims: sampledAttempts.length - readyAttempts.length, + timeoutPending: sampledAttempts.filter(isClawScanTimeout).length, + scannerFailurePending: sampledAttempts.filter( + (attempt) => attempt.checks.clawscan.status === "failed", + ).length, + oldestPendingAgeSeconds: oldestPendingAttempt + ? Math.max(0, Math.floor((snapshotAt - oldestPendingAttempt.createdAt) / 1000)) + : 0, + oldestReadyAgeSeconds: oldestReadyAttempt + ? Math.max(0, Math.floor((snapshotAt - oldestReadyAttempt.createdAt) / 1000)) + : 0, + }; + }, +}); + +export const logPrePublicationQueueHealthInternal = internalAction({ + args: {}, + handler: async (ctx): Promise => { + const snapshot = await runQueryRef( + ctx, + internalRefs.prepublicationObservability.getPrePublicationQueueHealthInternal, + {}, + ); + logEvent(Events.PrePublicationQueueSnapshot, snapshot); + return snapshot; + }, +});