feat(observability): log prepublication queue health (#3192)

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;

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");
 

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 },

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;
 

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<unknown>;
+  }
+)._handler;
+const logQueueHealthHandler = (
+  logPrePublicationQueueHealthInternal as unknown as {
+    _handler: (ctx: unknown, args: unknown) => Promise<unknown>;
+  }
+)._handler;
+
+function makeQueueHealthCtx(attempts: Array<Record<string, unknown>>) {
+  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<string, unknown>();
+              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,
+      }),
+    );
+  });
+});

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<T>(
+  ctx: { runQuery: (ref: never, args: never) => Promise<unknown> },
+  ref: unknown,
+  args: unknown,
+): Promise<T> {
+  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<PrePublicationQueueHealth> => {
+    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<PrePublicationQueueHealth> => {
+    const snapshot = await runQueryRef<PrePublicationQueueHealth>(
+      ctx,
+      internalRefs.prepublicationObservability.getPrePublicationQueueHealthInternal,
+      {},
+    );
+    logEvent(Events.PrePublicationQueueSnapshot, snapshot);
+    return snapshot;
+  },
+});