fix: improve sql queries

2026-04-09 04:48:50 +04:00
parent 89f0856a04
commit 91ca996e74
10 changed files with 495 additions and 219 deletions
@@ -0,0 +1,62 @@
+// In-memory cache of the enabled monitor list, keyed by region. The /internal/due
+// endpoint is polled by every runner roughly once a second per region; without
+// this cache each poll re-runs a 500-row scan against `monitors` with an array
+// predicate, which dominates the api's Postgres traffic at any real fleet size.
+//
+// The list almost never changes between polls — monitor create/edit/delete is at
+// most a few times per hour. So we memoize per-region with a short TTL and bust
+// the cache from the monitor mutation handlers so edits are visible instantly.
+
+import sql from "../db";
+
+const TTL_MS = 5000;
+
+type MonitorRow = Record<string, any>;
+type Entry = { rows: MonitorRow[]; expiresAt: number };
+
+const cache = new Map<string, Entry>();
+const inflight = new Map<string, Promise<MonitorRow[]>>();
+
+async function fetchForRegion(region: string): Promise<MonitorRow[]> {
+  return sql<MonitorRow[]>`
+    SELECT id, url, method, request_headers, request_body, timeout_ms, interval_s, query, regions,
+           max_retries, retry_interval_s, created_at
+    FROM monitors
+    WHERE enabled = true
+      AND (
+        array_length(regions, 1) IS NULL
+        OR regions = '{}'
+        OR ${region} = ANY(regions)
+      )
+    LIMIT 500
+  `;
+}
+
+export async function getMonitorsForRegion(region: string): Promise<MonitorRow[]> {
+  const now = Date.now();
+  const hit = cache.get(region);
+  if (hit && hit.expiresAt > now) return hit.rows;
+
+  // Coalesce concurrent refreshes for the same region so a thundering herd of
+  // runner polls doesn't fan out into N parallel SELECTs against an expired
+  // entry.
+  let pending = inflight.get(region);
+  if (!pending) {
+    pending = fetchForRegion(region)
+      .then((rows) => {
+        cache.set(region, { rows, expiresAt: Date.now() + TTL_MS });
+        return rows;
+      })
+      .finally(() => { inflight.delete(region); });
+    inflight.set(region, pending);
+  }
+  return pending;
+}
+
+// Called by monitor create/patch/delete/toggle handlers. Wipes the entire
+// region map — fine because (a) entries are tiny, (b) refresh is cheap, and
+// (c) we don't know which regions a freshly-edited monitor belongs to without
+// reading it back. Simpler than per-region invalidation, identical net effect.
+export function invalidateMonitorList(): void {
+  cache.clear();
+}
@@ -2,7 +2,7 @@ import { Elysia } from "elysia";
 import { ingest } from "./routes/pings";
 import { monitors } from "./routes/monitors";
 import { account } from "./routes/auth";
-import { internal } from "./routes/internal";
+import { internal, startPruneJob } from "./routes/internal";
 import { channels } from "./routes/channels";
 import { statusPages } from "./routes/status_pages";
 import { incidents } from "./routes/incidents";
@@ -21,6 +21,7 @@ process.on("uncaughtException", (err) => {

 await migrate();
 await startRollupJob();
+startPruneJob();

 const elysia = new Elysia()
  .get("/", () => ({
@@ -4,8 +4,13 @@ import sql from "../db";
 // widgets can compute uptime % over arbitrary windows without ever scanning the
 // pings table at read time. Two resolutions: hourly and daily.
 //
-// Each pass aggregates the *current* bucket only. The query is bounded by the
-// bucket size, not the table size, so it's cheap regardless of history depth.
+// Watermark model: rollup_watermarks(bucket_type) records the most recent
+// pings.checked_at that has been folded into this bucket_type's rollup rows.
+// Each periodic pass scans ONLY pings newer than the watermark, then merges
+// them into existing rollup rows additively (total = total + new_total, etc.)
+// via ON CONFLICT … DO UPDATE. This makes per-pass work proportional to the
+// delta of new pings, not the bucket size — critical once a single account has
+// thousands of monitors.

 type BucketType = "hourly" | "daily";

@@ -14,11 +19,49 @@ const BUCKET_TRUNC: Record<BucketType, string> = {
  daily:  "day",
 };

-async function rollupCurrent(bucket: BucketType): Promise<number> {
+// Pull the current watermark for a bucket type. First call after a fresh
+// deploy returns epoch (1970-01-01) so the first pass folds the entire
+// retention window in one go.
+async function getWatermark(bucket: BucketType): Promise<Date> {
+  const [row] = await sql<{ last_aggregated_at: Date }[]>`
+    SELECT last_aggregated_at FROM rollup_watermarks WHERE bucket_type = ${bucket}
+  `;
+  return row?.last_aggregated_at ?? new Date(0);
+}
+
+async function setWatermark(bucket: BucketType, ts: Date): Promise<void> {
+  await sql`
+    INSERT INTO rollup_watermarks (bucket_type, last_aggregated_at)
+    VALUES (${bucket}, ${ts})
+    ON CONFLICT (bucket_type) DO UPDATE SET last_aggregated_at = EXCLUDED.last_aggregated_at
+  `;
+}
+
+// Aggregate every ping newer than the watermark into the appropriate bucket
+// row. Merges into existing rows additively so we never re-scan rows that
+// were already folded in on a previous pass. The avg_latency merge is a
+// weighted average over the running totals.
+async function rollupSinceWatermark(bucket: BucketType): Promise<number> {
  const trunc = BUCKET_TRUNC[bucket];
+  const watermark = await getWatermark(bucket);
+
+  // Capture the upper bound BEFORE the aggregation runs so we don't miss any
+  // pings that arrive between SELECT and the watermark write. Anything ingested
+  // after `boundary` will be picked up on the next pass.
+  const boundary = new Date();
+
  // GROUP BY 1,2,4 (ordinals) instead of repeating the date_trunc expression —
  // when the unit is a $-bound parameter, Postgres won't recognize the two
  // expressions as identical and will reject the column. Ordinals are safe.
+  //
+  // The ON CONFLICT merge formula:
+  //   total       = old + new
+  //   up_count    = old + new
+  //   avg_latency = weighted average of old and new, weighted by their totals
+  //                 (NULLIF guards against the degenerate empty-bucket case)
+  // The weighted-average formula is mathematically identical to a one-shot
+  // SUM(latency)/SUM(total) recompute, so the merged value is byte-equal to
+  // what a full rescan would produce.
  const result = await sql`
    INSERT INTO monitor_uptime_rollup (monitor_id, region, bucket_type, bucket_start, total, up_count, avg_latency)
    SELECT
@@ -30,22 +73,31 @@ async function rollupCurrent(bucket: BucketType): Promise<number> {
      count(*) FILTER (WHERE up)::int AS up_count,
      avg(latency_ms)::real AS avg_latency
    FROM pings
-    WHERE checked_at >= date_trunc(${trunc}, now())
+    WHERE checked_at > ${watermark} AND checked_at <= ${boundary}
    GROUP BY 1, 2, 3, 4
    ON CONFLICT (monitor_id, region, bucket_type, bucket_start) DO UPDATE SET
-      total       = EXCLUDED.total,
-      up_count    = EXCLUDED.up_count,
-      avg_latency = EXCLUDED.avg_latency
+      total       = monitor_uptime_rollup.total + EXCLUDED.total,
+      up_count    = monitor_uptime_rollup.up_count + EXCLUDED.up_count,
+      avg_latency = (
+        COALESCE(monitor_uptime_rollup.avg_latency, 0) * monitor_uptime_rollup.total
+        + COALESCE(EXCLUDED.avg_latency, 0) * EXCLUDED.total
+      ) / NULLIF(monitor_uptime_rollup.total + EXCLUDED.total, 0)
  `;
+
+  await setWatermark(bucket, boundary);
  return result.count ?? 0;
 }

-// Walk back N units and aggregate any buckets that don't exist yet. Used at
-// startup so a freshly-deployed system has historical data immediately.
-async function backfillRecent(bucket: BucketType, units: number): Promise<number> {
+// One-shot recompute over an arbitrary window, fully overwriting matched rows.
+// Used for the startup backfill and the "still empty after backfill" force-run.
+// Takes an explicit upper boundary so the caller can capture it BEFORE running
+// the recompute and use the same value for the watermark write afterwards —
+// any ping with checked_at > boundary is guaranteed to be outside this window
+// and will be picked up by the first incremental pass instead. This closes
+// the race where a ping ingested between boundary capture and the SELECT
+// could otherwise be folded in by both recompute and incremental.
+async function recomputeWindow(bucket: BucketType, units: number, boundary: Date): Promise<number> {
  const trunc = BUCKET_TRUNC[bucket];
-  // Build the interval string entirely in JS so postgres.js binds a single text
-  // parameter. Avoids the int || text type-mismatch trap inside SQL.
  const intervalLiteral = `${units} ${trunc}s`;
  const result = await sql`
    INSERT INTO monitor_uptime_rollup (monitor_id, region, bucket_type, bucket_start, total, up_count, avg_latency)
@@ -58,9 +110,13 @@ async function backfillRecent(bucket: BucketType, units: number): Promise<number
      count(*) FILTER (WHERE up)::int AS up_count,
      avg(latency_ms)::real AS avg_latency
    FROM pings
-    WHERE checked_at >= date_trunc(${trunc}, now()) - ${intervalLiteral}::interval
+    WHERE checked_at >= date_trunc(${trunc}, ${boundary}::timestamptz) - ${intervalLiteral}::interval
+      AND checked_at <= ${boundary}
    GROUP BY 1, 2, 3, 4
-    ON CONFLICT (monitor_id, region, bucket_type, bucket_start) DO NOTHING
+    ON CONFLICT (monitor_id, region, bucket_type, bucket_start) DO UPDATE SET
+      total       = EXCLUDED.total,
+      up_count    = EXCLUDED.up_count,
+      avg_latency = EXCLUDED.avg_latency
  `;
  return result.count ?? 0;
 }
@@ -78,36 +134,45 @@ export async function startRollupJob() {
  if (started) return;
  started = true;

-  // Startup backfill. Errors are logged BUT NOT swallowed silently anymore —
-  // we throw so a broken rollup query trips the api process and shows in the
-  // service logs immediately, instead of leaving the table mysteriously empty.
+  // Startup backfill. Capture the boundary FIRST, then run the one-shot
+  // recompute bounded by it. The watermark is then set to the same boundary,
+  // so any ping with checked_at > boundary is guaranteed to be picked up by
+  // the first incremental pass — never folded in twice and never missed.
  try {
+    const boundary = new Date();
    const [h, d] = await Promise.all([
-      backfillRecent("hourly", 48),
-      backfillRecent("daily",  90),
+      recomputeWindow("hourly", 48, boundary),
+      recomputeWindow("daily",  90, boundary),
    ]);
    console.log(`[rollup] backfilled rows: hourly=${h} daily=${d}`);
+    await Promise.all([
+      setWatermark("hourly", boundary),
+      setWatermark("daily",  boundary),
+    ]);
  } catch (e) {
    console.error("[rollup] backfill FAILED — rollup table will be empty until fixed:", e);
  }

  // Force-run check: if any bucket type is still empty after the backfill,
-  // run rollupCurrent immediately so we have at least one row per type. This
-  // covers the "fresh deploy with very recent pings only" case.
+  // run an incremental pass immediately so we have at least one row per type.
+  // Covers the "fresh deploy with very recent pings only" case.
  try {
    for (const b of ["hourly", "daily"] as BucketType[]) {
      if (await rollupIsEmpty(b)) {
-        console.log(`[rollup] ${b} still empty — forcing current-bucket aggregation`);
-        await rollupCurrent(b);
+        console.log(`[rollup] ${b} still empty — forcing incremental aggregation`);
+        // Reset watermark so the pass picks up everything in retention.
+        await setWatermark(b, new Date(0));
+        await rollupSinceWatermark(b);
      }
    }
  } catch (e) {
    console.error("[rollup] force-run check failed:", e);
  }

-  // Periodic refreshes for the *current* bucket of each resolution. Each query
-  // is bounded by the current bucket only (date_trunc(...)) so it stays cheap
-  // even at high cadence.
-  setInterval(() => { rollupCurrent("hourly").catch((e) => console.warn("[rollup] hourly failed:", e)); },  30 * 1000);     // every 30s
-  setInterval(() => { rollupCurrent("daily").catch((e) => console.warn("[rollup] daily failed:",  e)); },  5 * 60 * 1000);  // every 5min
+  // Periodic incremental refreshes. Each pass scans only pings newer than the
+  // last watermark, so the work is proportional to the delta — not the bucket
+  // size. Hourly runs frequently so the current-hour bar appears quickly for
+  // fresh monitors; daily can run less often.
+  setInterval(() => { rollupSinceWatermark("hourly").catch((e) => console.warn("[rollup] hourly failed:", e)); },  30 * 1000);     // every 30s
+  setInterval(() => { rollupSinceWatermark("daily").catch((e) => console.warn("[rollup] daily failed:",  e)); },  5 * 60 * 1000);  // every 5min
 }
@@ -1,16 +1,91 @@
 import { Elysia } from "elysia";
 import sql from "../db";
 import { safeTokenCompare } from "../../../shared/auth";
+import { getMonitorsForRegion } from "../cache/monitor-list";

+// Chunked retention prune. A single big DELETE on the pings table holds locks
+// for the duration of the scan, blocks autovacuum from reclaiming dead tuples,
+// and can stall replication. We loop in 10k-row batches so each commit
+// releases its locks and lets autovacuum keep up. Total wall-clock is similar
+// to a one-shot DELETE but the per-statement impact on the rest of the system
+// is dramatically lower.
+const PRUNE_BATCH_SIZE = 10_000;
 export async function pruneOldPings(retentionDays = 90) {
-  const result = await sql`DELETE FROM pings WHERE checked_at < now() - ${retentionDays + ' days'}::interval`;
-  return result.count;
+  const interval = `${retentionDays} days`;
+  let total = 0;
+  while (true) {
+    const result = await sql`
+      WITH victims AS (
+        SELECT id FROM pings
+        WHERE checked_at < now() - ${interval}::interval
+        LIMIT ${PRUNE_BATCH_SIZE}
+      )
+      DELETE FROM pings WHERE id IN (SELECT id FROM victims)
+    `;
+    const batch = result.count ?? 0;
+    total += batch;
+    if (batch < PRUNE_BATCH_SIZE) break;
+  }
+  return total;
 }

-setInterval(() => {
-  const days = Number(process.env.PING_RETENTION_DAYS ?? 90);
-  pruneOldPings(days).catch((err) => console.error("Retention cleanup failed:", err));
-}, 60 * 60 * 1000);
+// Periodic prune is gated behind a Postgres session-level advisory lock so
+// horizontally scaled api replicas don't race each other. Without the lock,
+// two replicas running the same chunked DELETE would interleave their
+// LIMIT 10000 batches and compete for row locks. The lock is acquired with
+// pg_try_advisory_lock so a busy replica skips its tick instead of blocking.
+//
+// Session-level locks live on a specific Postgres backend connection, so we
+// reserve a connection from the pool with sql.reserve() and run the lock,
+// the chunked prune, and the unlock all on it. Releasing the reservation
+// returns the connection to the pool. If the api process crashes mid-prune,
+// the backend dies with it and Postgres releases the lock automatically.
+//
+// 134678338 is just an arbitrary lock id unique within this app. If we add
+// more global jobs later, give each its own constant in this file.
+const PRUNE_LOCK_ID = 134678338;
+let pruneJobStarted = false;
+
+async function runPruneTickWithLock(retentionDays: number): Promise<void> {
+  const reserved = await (sql as any).reserve();
+  try {
+    const [{ locked }] = await reserved`
+      SELECT pg_try_advisory_lock(${PRUNE_LOCK_ID}) AS locked
+    `;
+    if (!locked) return; // another replica is pruning right now
+    try {
+      const interval = `${retentionDays} days`;
+      let total = 0;
+      while (true) {
+        const result = await reserved`
+          WITH victims AS (
+            SELECT id FROM pings
+            WHERE checked_at < now() - ${interval}::interval
+            LIMIT ${PRUNE_BATCH_SIZE}
+          )
+          DELETE FROM pings WHERE id IN (SELECT id FROM victims)
+        `;
+        const batch = result.count ?? 0;
+        total += batch;
+        if (batch < PRUNE_BATCH_SIZE) break;
+      }
+      if (total > 0) console.log(`[prune] retention pruned ${total} pings (>${retentionDays}d)`);
+    } finally {
+      await reserved`SELECT pg_advisory_unlock(${PRUNE_LOCK_ID})`;
+    }
+  } finally {
+    reserved.release();
+  }
+}
+
+export function startPruneJob() {
+  if (pruneJobStarted) return;
+  pruneJobStarted = true;
+  setInterval(() => {
+    const days = Number(process.env.PING_RETENTION_DAYS ?? 90);
+    runPruneTickWithLock(days).catch((err) => console.error("Retention cleanup failed:", err));
+  }, 60 * 60 * 1000);
+}

 export const internal = new Elysia({ prefix: "/internal", detail: { hide: true } })
  .derive(({ headers, set }) => {
@@ -37,18 +112,13 @@ export const internal = new Elysia({ prefix: "/internal", detail: { hide: true }
    // creation is a tick. We pull all enabled monitors that match this
    // region, compute the next tick in JS, and return the ones whose next
    // tick falls within the lookahead window.
-    const monitors = await sql`
-      SELECT id, url, method, request_headers, request_body, timeout_ms, interval_s, query, regions,
-             max_retries, retry_interval_s, created_at
-      FROM monitors
-      WHERE enabled = true
-        AND (
-          array_length(regions, 1) IS NULL
-          OR regions = '{}'
-          OR ${region} = ANY(regions)
-        )
-      LIMIT 500
-    `;
+    //
+    // The monitor list itself is memoized in apps/api/src/cache/monitor-list.ts
+    // with a 5s TTL — runners poll this endpoint roughly once a second per
+    // region, but the underlying list almost never changes between polls. The
+    // cache is busted from monitor create/patch/delete/toggle so edits show up
+    // immediately.
+    const monitors = await getMonitorsForRegion(region);

    const nowMs = Date.now();
    const lookaheadEnd = nowMs + lookaheadMs;
@@ -3,6 +3,7 @@ import { requireAuth } from "./auth";
 import sql from "../db";
 import { validateMonitorUrl } from "../utils/ssrf";
 import { getPlanLimits } from "../../../shared/plans";
+import { invalidateMonitorList } from "../cache/monitor-list";

 const MonitorBody = t.Object({
  name:            t.String({ maxLength: 200, description: "Human-readable name" }),
@@ -111,6 +112,7 @@ export const monitors = new Elysia({ prefix: "/monitors" })
    `;
    if (body.channel_ids) await replaceMonitorChannels(monitor.id, accountId, body.channel_ids);
    if (body.tags) await replaceMonitorTags(monitor.id, body.tags);
+    invalidateMonitorList();
    return monitor;
  }, { body: MonitorBody, detail: { summary: "Create monitor", tags: ["monitors"] } })

@@ -177,6 +179,7 @@ export const monitors = new Elysia({ prefix: "/monitors" })
    if (!monitor) { set.status = 404; return { error: "Not found" }; }
    if (body.channel_ids) await replaceMonitorChannels(monitor.id, accountId, body.channel_ids);
    if (body.tags) await replaceMonitorTags(monitor.id, body.tags);
+    invalidateMonitorList();
    return monitor;
  }, { body: t.Partial(MonitorBody), detail: { summary: "Update monitor", tags: ["monitors"] } })

@@ -185,6 +188,7 @@ export const monitors = new Elysia({ prefix: "/monitors" })
      DELETE FROM monitors WHERE id = ${params.id} AND account_id = ${accountId} RETURNING id
    `;
    if (!deleted) { set.status = 404; return { error: "Not found" }; }
+    invalidateMonitorList();
    return { deleted: true };
  }, { detail: { summary: "Delete monitor", tags: ["monitors"] } })

@@ -195,6 +199,7 @@ export const monitors = new Elysia({ prefix: "/monitors" })
      RETURNING id, enabled
    `;
    if (!monitor) { set.status = 404; return { error: "Not found" }; }
+    invalidateMonitorList();
    return monitor;
  }, { detail: { summary: "Toggle monitor on/off", tags: ["monitors"] } })

@@ -56,10 +56,28 @@ export const ingest = new Elysia()
    const token = headers["x-monitor-token"];
    if (!safeTokenCompare(token, process.env.MONITOR_TOKEN)) { set.status = 401; return { error: "Unauthorized" }; }

-    const [monitor_check] = await sql`
-      SELECT id, account_id, name, url, resend_interval, cert_alert_days
-      FROM monitors WHERE id = ${body.monitor_id}
-    `;
+    // Per-region transition state. Region is always populated by current runners;
+    // legacy null values from older pings collapse to "default" so state and
+    // notifications never carry an empty label.
+    const region = body.region && body.region.length > 0 ? body.region : 'default';
+
+    // The monitor lookup and the per-region state lookup are independent —
+    // the state row's primary key doesn't depend on anything from the monitor
+    // row. Fire them in parallel to halve the wall-clock cost on the hottest
+    // path in the system. (Combining them into a JOIN is a wash on a warm
+    // pool: both sides are PK lookups, and a JOIN just adds nested-loop
+    // planner overhead. Promise.all keeps each query's plan trivial.)
+    const [[monitor_check], [stateRow]] = await Promise.all([
+      sql`
+        SELECT id, account_id, name, url, resend_interval, cert_alert_days
+        FROM monitors WHERE id = ${body.monitor_id}
+      `,
+      sql`
+        SELECT last_state, consecutive_down, cert_alert_sent
+        FROM monitor_region_state
+        WHERE monitor_id = ${body.monitor_id} AND region = ${region}
+      `,
+    ]);
    if (!monitor_check) { set.status = 404; return { error: "Monitor not found" }; }

    const meta = body.meta ? { ...body.meta } : {};
@@ -72,16 +90,6 @@ export const ingest = new Elysia()
    const scheduledAt = body.scheduled_at ? new Date(body.scheduled_at) : null;
    const jitterMs = body.jitter_ms ?? null;

-    // Per-region transition state. Region is always populated by current runners;
-    // legacy null values from older pings collapse to "default" so state and
-    // notifications never carry an empty label.
-    const region = body.region && body.region.length > 0 ? body.region : 'default';
-    const [stateRow] = await sql`
-      SELECT last_state, consecutive_down, cert_alert_sent
-      FROM monitor_region_state
-      WHERE monitor_id = ${body.monitor_id} AND region = ${region}
-    `;
-
    const newState = body.up ? 'up' : 'down';
    const prevState: string | null = stateRow?.last_state ?? null;
    let consecutiveDown: number = stateRow?.consecutive_down ?? 0;
@@ -62,17 +62,22 @@ async function replaceGroupsAndMonitors(
  if (groups !== undefined) {
    await sql`DELETE FROM status_page_groups WHERE status_page_id = ${pageId}`;
  }
+  // Single bulk INSERT instead of one round-trip per group. The RETURNING set
+  // comes back in INSERT order, which equals the array order — that lets us
+  // map index → id without a follow-up SELECT. Mirrors the bulk insert pattern
+  // used by the monitors block right below.
  const groupIds: string[] = [];
  if (groups && groups.length > 0) {
-    for (let i = 0; i < groups.length; i++) {
-      const g = groups[i]!;
-      const [row] = await sql<{ id: string }[]>`
-        INSERT INTO status_page_groups (status_page_id, name, position)
-        VALUES (${pageId}, ${g.name}, ${g.position ?? i})
-        RETURNING id
-      `;
-      groupIds.push(row!.id);
-    }
+    const rows = groups.map((g, i) => ({
+      status_page_id: pageId,
+      name:           g.name,
+      position:       g.position ?? i,
+    }));
+    const inserted = await sql<{ id: string }[]>`
+      INSERT INTO status_page_groups ${sql(rows, "status_page_id", "name", "position")}
+      RETURNING id
+    `;
+    for (const r of inserted) groupIds.push(r.id);
  }

  if (monitorsList !== undefined) {