feat: KEEP-1042 per-workflow execution rate gauge

lib/metrics/METRICS_REFERENCE.md

@@ -72,6 +72,7 @@ Counter/Gauge metrics tracking request/event counts.
 |-------------|-------------|--------|------|--------|
 | `workflow.executions.total` | Total workflow executions by status (all-time) | `status`, `org_slug`, `error_type` (`user`/`system`/`unknown`/`na`) | gauge | DB |
 | `workflow.execution.errors.total` | Total failed workflow executions (all-time) | - | gauge | DB |
+| `workflow.executions.last_hour` | Runs started in the **last hour** for the busiest workflows (top 20 by runs, plus top 20 by errored runs), as `outcome="all"` and `outcome="errored"`. Powers the per-workflow execution rate alerts, which read it directly. Not cumulative. | `workflow_id`, `org_slug`, `outcome` | gauge | DB |
 | `plugin.invocations.total` | Plugin action invocations | `plugin_name`, `action_name` | count | API |
 | `user.active.daily` | Daily active users (24h) | - | gauge | DB |
 

lib/metrics/collectors/prometheus.ts

@@ -161,6 +161,19 @@ const workflowErrorsByWorkflow = getOrCreateGauge(
   ["workflow_id", "org_slug", "error_type"]
 );
 
+// Runs started in the last hour for the busiest workflows (see
+// getWorkflowExecutionRatesFromDb), as outcome="all" and outcome="errored".
+// The per-workflow execution rate alert reads it directly: a runaway block or
+// event trigger shows up as one workflow_id far above the rest. Cardinality is
+// bounded by the one-hour window and the top-N pick, at most 2 x N workflows
+// with two series each. No `_total` suffix: it is a poll-driven gauge.
+const workflowExecutionsLastHour = getOrCreateGauge(
+  dbRegistry,
+  "keeperhub_workflow_executions_last_hour",
+  "Runs started in the last hour for the busiest workflows, by workflow_id, org_slug and outcome (all or errored)",
+  ["workflow_id", "org_slug", "outcome"]
+);
+
 // TECH-6544: errored executions in the last hour, PLATFORM-WIDE, grouped by
 // (error_category, error_type). Keyed on error_category so the infra P3 alert
 // dedups system errors by *cause* — one series per failure mode — rather than
@@ -1909,6 +1922,7 @@ async function refreshDbMetricsNow(): Promise<void> {
       getExecutionRetentionStatsFromDb,
       getStuckPendingTransactionCountsFromDb,
       getWorkflowErrorsByWorkflowFromDb,
+      getWorkflowExecutionRatesFromDb,
       getSystemErrorsByCategoryFromDb,
       getStepStatsFromDb,
       getDailyActiveUsersFromDb,
@@ -1930,6 +1944,7 @@ async function refreshDbMetricsNow(): Promise<void> {
       retentionStats,
       stuckPendingTxCounts,
       errorsByWorkflow,
+      executionRates,
       systemErrorsByCategoryRows,
       stepStats,
       dailyActiveUsers,
@@ -1950,6 +1965,7 @@ async function refreshDbMetricsNow(): Promise<void> {
       getExecutionRetentionStatsFromDb(),
       getStuckPendingTransactionCountsFromDb(),
       getWorkflowErrorsByWorkflowFromDb(),
+      getWorkflowExecutionRatesFromDb(),
       getSystemErrorsByCategoryFromDb(),
       getStepStatsFromDb(),
       getDailyActiveUsersFromDb(),
@@ -2046,6 +2062,18 @@ async function refreshDbMetricsNow(): Promise<void> {
       );
     }
 
+    // Reset before populating so a workflow that drops out of the top list, or
+    // stops running, clears out instead of pinning its last value.
+    workflowExecutionsLastHour.reset();
+    for (const row of executionRates) {
+      const labels = { workflow_id: row.workflowId, org_slug: row.orgSlug };
+      workflowExecutionsLastHour.set({ ...labels, outcome: "all" }, row.runs);
+      workflowExecutionsLastHour.set(
+        { ...labels, outcome: "errored" },
+        row.errored
+      );
+    }
+
     // TECH-6544: errored executions per (error_category, error_type),
     // platform-wide. Reset before populating so a category that no longer has
     // errors in the window clears out instead of pinning a stale value.

lib/metrics/db-metrics.ts

@@ -594,6 +594,90 @@ export async function getWorkflowErrorsByWorkflowFromDb(): Promise<WorkflowError
   }
 }
 
+/**
+ * Runs started in the last hour, per workflow, and how many of them errored.
+ *
+ * Feeds the per-workflow execution rate alert. A block or event trigger can
+ * start a run for every block the chain produces, and nothing bounded it: one
+ * such trigger once wrote over a million runs before anyone noticed the table.
+ * A one-hour window catches that within hours.
+ *
+ * The window is a range scan on idx_workflow_executions_started_at. Only the
+ * busiest workflows are kept (see pickWorkflowExecutionRates), so the gauge
+ * stays small however many workflows run.
+ */
+export type WorkflowExecutionRate = {
+  workflowId: string;
+  orgSlug: string;
+  runs: number;
+  errored: number;
+};
+
+export const WORKFLOW_EXECUTION_RATE_TOP_N = 20;
+
+/**
+ * The top workflows by runs, plus the top workflows by errored runs. A workflow
+ * that errors on every run can sit below the busiest ones by volume, and error
+ * accumulation is half of what the alert is for.
+ */
+export function pickWorkflowExecutionRates(
+  rows: WorkflowExecutionRate[],
+  topN: number = WORKFLOW_EXECUTION_RATE_TOP_N
+): WorkflowExecutionRate[] {
+  const byRuns = [...rows].sort((a, b) => b.runs - a.runs).slice(0, topN);
+  const byErrored = rows
+    .filter((row) => row.errored > 0)
+    .sort((a, b) => b.errored - a.errored)
+    .slice(0, topN);
+  const picked = new Map<string, WorkflowExecutionRate>();
+  for (const row of [...byRuns, ...byErrored]) {
+    picked.set(row.workflowId, row);
+  }
+  return [...picked.values()];
+}
+
+/**
+ * The query behind getWorkflowExecutionRatesFromDb. Exported so a test can
+ * EXPLAIN the SQL the collector really sends, not a hand-written copy of it.
+ */
+export function workflowExecutionRatesQuery() {
+  return db
+    .select({
+      workflowId: workflowExecutions.workflowId,
+      orgSlug: sql<string>`COALESCE(${organization.slug}, 'none')`,
+      runs: count(),
+      errored: sql<string>`COUNT(*) FILTER (WHERE ${inArray(workflowExecutions.status, [...ERROR_STATUSES])})`,
+    })
+    .from(workflowExecutions)
+    .innerJoin(workflows, eq(workflowExecutions.workflowId, workflows.id))
+    .leftJoin(organization, eq(workflows.organizationId, organization.id))
+    .where(sql`${workflowExecutions.startedAt} >= now() - interval '1 hour'`)
+    .groupBy(workflowExecutions.workflowId, organization.slug);
+}
+
+export async function getWorkflowExecutionRatesFromDb(): Promise<
+  WorkflowExecutionRate[]
+> {
+  try {
+    const rows = await workflowExecutionRatesQuery();
+    return pickWorkflowExecutionRates(
+      rows.map((row) => ({
+        workflowId: row.workflowId,
+        orgSlug: row.orgSlug,
+        runs: Number(row.runs) || 0,
+        errored: Number(row.errored) || 0,
+      }))
+    );
+  } catch (error) {
+    logSystemWarn(
+      ErrorCategory.DATABASE,
+      "[Metrics] Failed to query per-workflow execution rates from DB",
+      error
+    );
+    return [];
+  }
+}
+
 // TECH-6544: per-(error_category, error_type) error counts over a ROLLING
 // 1-HOUR window, PLATFORM-WIDE (all orgs). System errors are platform faults
 // (DB, RPC, infra, workflow engine), not a managed-client concern, so this

tests/db/workflow-execution-rates.db.test.ts

@@ -0,0 +1,147 @@
+/**
+ * The per-workflow execution rate collector against a real Postgres.
+ *
+ * It runs on every metrics scrape, on the metrics pool and under its statement
+ * timeout, so two things matter beyond the numbers: the one-hour window must be
+ * answerable from idx_workflow_executions_started_at rather than a scan of the
+ * whole run table, and the FILTER on the errored statuses must count what the
+ * alert thinks it counts.
+ */
+
+import "dotenv/config";
+import { drizzle } from "drizzle-orm/postgres-js";
+import postgres from "postgres";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import {
+  organization,
+  users,
+  workflowExecutions,
+  workflows,
+} from "../../lib/db/schema";
+
+// vitest runs in Node, not an SSR context; the metrics module is server-only.
+vi.mock("server-only", () => ({}));
+// tests/setup.ts globally stubs @/lib/db. The whole point here is the SQL.
+vi.unmock("@/lib/db");
+
+const DATABASE_URL = process.env.DATABASE_URL ?? "";
+
+const PREFIX = "test_rates_";
+const USER = `${PREFIX}user`;
+const ORG = `${PREFIX}org`;
+const WF_BUSY = `${PREFIX}wf_busy`;
+const WF_BAD = `${PREFIX}wf_bad`;
+const WF_OLD = `${PREFIX}wf_old`;
+const MINUTE_MS = 60 * 1000;
+
+type Metrics = typeof import("@/lib/metrics/db-metrics");
+
+describe("per-workflow execution rates (real database)", () => {
+  let queryClient: ReturnType<typeof postgres>;
+  let db: ReturnType<typeof drizzle>;
+  let metrics: Metrics;
+
+  async function cleanup(): Promise<void> {
+    const like = `${PREFIX}%`;
+    await queryClient`DELETE FROM workflow_executions WHERE id LIKE ${like}`;
+    await queryClient`DELETE FROM workflows WHERE id LIKE ${like}`;
+    await queryClient`DELETE FROM organization WHERE id LIKE ${like}`;
+    await queryClient`DELETE FROM users WHERE id LIKE ${like}`;
+  }
+
+  beforeAll(async () => {
+    // Seeds organizations into a database the retention suite counts in full.
+    const host = new URL(DATABASE_URL).hostname;
+    if (!["localhost", "127.0.0.1", "::1", "postgres", "db"].includes(host)) {
+      throw new Error(`refusing to run against a non-local database: ${host}`);
+    }
+    queryClient = postgres(DATABASE_URL);
+    db = drizzle(queryClient);
+    metrics = await import("@/lib/metrics/db-metrics");
+
+    await cleanup();
+    // The window is now() - 1 hour inside the query, so the seed is relative to
+    // the wall clock rather than to a fixed instant.
+    const now = Date.now();
+    const minutesAgo = (minutes: number): Date =>
+      new Date(now - minutes * MINUTE_MS);
+
+    await db.insert(users).values({
+      id: USER,
+      name: "rates probe",
+      email: `${PREFIX}probe@keeperhub.test`,
+      emailVerified: true,
+      createdAt: new Date(now),
+      updatedAt: new Date(now),
+    });
+    await db
+      .insert(organization)
+      .values({ id: ORG, name: ORG, slug: ORG, createdAt: new Date(now) });
+    await db.insert(workflows).values(
+      [WF_BUSY, WF_BAD, WF_OLD].map((id) => ({
+        id,
+        name: id,
+        userId: USER,
+        organizationId: ORG,
+        nodes: [],
+        edges: [],
+      }))
+    );
+
+    const run = (
+      workflowId: string,
+      n: number,
+      status: "success" | "error" | "system_error",
+      startedAt: Date
+    ) => ({
+      id: `${workflowId}_run_${status}_${n}`,
+      workflowId,
+      userId: USER,
+      status,
+      startedAt,
+    });
+    await db.insert(workflowExecutions).values([
+      ...[1, 2, 3, 4, 5].map((n) => run(WF_BUSY, n, "success", minutesAgo(10))),
+      run(WF_BAD, 1, "error", minutesAgo(20)),
+      run(WF_BAD, 2, "system_error", minutesAgo(30)),
+      run(WF_BAD, 3, "success", minutesAgo(40)),
+      // Outside the window: two hours old, however many there are.
+      ...[1, 2, 3, 4, 5, 6, 7].map((n) =>
+        run(WF_OLD, n, "error", minutesAgo(120))
+      ),
+    ]);
+  });
+
+  afterAll(async () => {
+    await cleanup();
+    await queryClient.end();
+  });
+
+  it("counts runs and errored runs per workflow inside the last hour only", async () => {
+    const rates = await metrics.getWorkflowExecutionRatesFromDb();
+    const mine = rates
+      .filter((rate) => rate.workflowId.startsWith(PREFIX))
+      .sort((a, b) => a.workflowId.localeCompare(b.workflowId));
+
+    expect(mine).toEqual([
+      { workflowId: WF_BAD, orgSlug: ORG, runs: 3, errored: 2 },
+      { workflowId: WF_BUSY, orgSlug: ORG, runs: 5, errored: 0 },
+    ]);
+  });
+
+  it("answers the window from idx_workflow_executions_started_at", async () => {
+    const query = metrics.workflowExecutionRatesQuery().toSQL();
+    const plan = await queryClient.begin(async (tx) => {
+      // A handful of seeded rows makes a sequential scan the cheapest plan,
+      // so it is turned off to ask whether the index can answer at all.
+      await tx`SET LOCAL enable_seqscan = off`;
+      return await tx.unsafe(
+        `EXPLAIN (FORMAT JSON) ${query.sql}`,
+        query.params as never[]
+      );
+    });
+    expect(JSON.stringify(plan)).toContain(
+      '"Index Name":"idx_workflow_executions_started_at"'
+    );
+  });
+});

tests/unit/db-metrics-cache.test.ts

@@ -35,6 +35,7 @@ const DEFAULT_DB_RETURNS: Record<string, unknown> = {
   },
   getStuckPendingTransactionCountsFromDb: [],
   getWorkflowErrorsByWorkflowFromDb: [],
+  getWorkflowExecutionRatesFromDb: [],
   getSystemErrorsByCategoryFromDb: [],
   getStepStatsFromDb: {
     countsByType: {},
@@ -153,6 +154,10 @@ const ERRORS_BY_WORKFLOW_SKY_RE =
   /keeperhub_workflow_errors_by_workflow\{[^}]*workflow_id="wf_sky_1"[^}]*org_slug="techops-services"[^}]*error_type="user"[^}]*\}\s+7/;
 const ERRORS_BY_WORKFLOW_AJNA_RE =
   /keeperhub_workflow_errors_by_workflow\{[^}]*workflow_id="wf_ajna_1"[^}]*org_slug="ajna"[^}]*error_type="system"[^}]*\}\s+2/;
+const EXECUTIONS_LAST_HOUR_ALL_RE =
+  /keeperhub_workflow_executions_last_hour\{[^}]*workflow_id="wf_runaway"[^}]*org_slug="acme"[^}]*outcome="all"[^}]*\}\s+4200/;
+const EXECUTIONS_LAST_HOUR_ERRORED_RE =
+  /keeperhub_workflow_executions_last_hour\{[^}]*workflow_id="wf_runaway"[^}]*org_slug="acme"[^}]*outcome="errored"[^}]*\}\s+3900/;
 const ERRORS_BY_CATEGORY_SYSTEM_RE =
   /keeperhub_system_errors_by_category\{[^}]*error_category="network_rpc"[^}]*error_type="system"[^}]*\}\s+5/;
 const ERRORS_BY_CATEGORY_UNKNOWN_RE =
@@ -466,6 +471,51 @@ describe("keeperhub_workflow_errors_by_workflow gauge", () => {
   });
 });
 
+describe("keeperhub_workflow_executions_last_hour gauge", () => {
+  const originalTtl = process.env.DB_METRICS_CACHE_TTL_MS;
+
+  beforeEach(() => {
+    __resetDbMetricsCacheForTest();
+    for (const fn of Object.values(dbMocks)) {
+      fn.mockReset();
+    }
+    rebindDefaultDbMockImplementations();
+    process.env.DB_METRICS_CACHE_TTL_MS = "0";
+  });
+
+  afterEach(() => {
+    if (originalTtl === undefined) {
+      delete process.env.DB_METRICS_CACHE_TTL_MS;
+    } else {
+      process.env.DB_METRICS_CACHE_TTL_MS = originalTtl;
+    }
+  });
+
+  it("emits an all series and an errored series per workflow", async () => {
+    dbMocks.getWorkflowExecutionRatesFromDb.mockResolvedValue([
+      { workflowId: "wf_runaway", orgSlug: "acme", runs: 4200, errored: 3900 },
+    ]);
+
+    await updateDbMetrics();
+    const out = await getDbMetrics();
+
+    expect(out).toMatch(EXECUTIONS_LAST_HOUR_ALL_RE);
+    expect(out).toMatch(EXECUTIONS_LAST_HOUR_ERRORED_RE);
+  });
+
+  it("clears a workflow that drops out of the query", async () => {
+    dbMocks.getWorkflowExecutionRatesFromDb.mockResolvedValueOnce([
+      { workflowId: "wf_gone", orgSlug: "acme", runs: 10, errored: 0 },
+    ]);
+    await updateDbMetrics();
+    expect(await getDbMetrics()).toContain('workflow_id="wf_gone"');
+
+    dbMocks.getWorkflowExecutionRatesFromDb.mockResolvedValue([]);
+    await updateDbMetrics();
+    expect(await getDbMetrics()).not.toContain('workflow_id="wf_gone"');
+  });
+});
+
 // TECH-6544: the system-errors-by-category gauge dedups errors by cause for the
 // infra P3 alert. It must be DB-sourced and populated on the metrics scrape,
 // grouped by (error_category, error_type) only — platform-wide, with no