Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/openstatusHQ/openstatus. 🫖 Status page with uptime monitoring & API monitoring as code 🫖 openstatus.dev
bun drizzle-orm monitoring monitoring-as-code nextjs observability on-call open-source shadcn-ui status-page statuspage synthetic-monitoring tinybird turso uptime uptime-checker uptime-monitor
Something went wrong. Try again.
21 kB · 717 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718import { getLogger } from "@logtape/logtape";import { and, db, eq, inArray, lt, notInArray, sql } from "@openstatus/db";import type { MonitorIncident, Monitor, Notification,} from "@openstatus/db/src/schema";import { notificationOutbox, monitorIncidentTable, monitor, notification, notificationDeadLetter, notificationTrigger, selectMonitorSchema, selectNotificationSchema,} from "@openstatus/db/src/schema";import { withBusyRetry } from "@openstatus/services";import * as Sentry from "@sentry/deno";import { Effect, Exit, Queue } from "effect";
import { env } from "../env";import { checkerAudit } from "../utils/audit-log";import { loadSmsQuotaBlocked } from "./sms-quota";import { providerToFunction } from "./utils";
const logger = getLogger(["workflow"]);
const CLAIM_LIMIT = 20;const DELIVERY_CONCURRENCY = 5;const MAX_BACKOFF_MS = 30_000;
/** * A claim leases a row for one delivery attempt, not for the whole message: a * worker that dies mid-send has to hand the row back long before the deadline. * The lease still has to outlast the worst case for a claimed row, which is * waiting behind a full batch at the delivery concurrency, plus the commit. */const LEASE_SLACK_MS = 5_000;
function leaseSeconds(limit: number, timeoutMs: number): number { const waves = Math.ceil(limit / DELIVERY_CONCURRENCY); return Math.ceil((waves * timeoutMs + LEASE_SLACK_MS) / 1000);}
const workerId = crypto.randomUUID();
type OutboxRow = typeof notificationOutbox.$inferSelect;
type DeliveryOutcome = | { kind: "delivered"; row: OutboxRow } | { kind: "skipped"; row: OutboxRow; reason: string } | { kind: "released"; row: OutboxRow } | { kind: "retry"; row: OutboxRow; error: string; delayMs: number; nextAttemptAt: number; } | { kind: "dead"; row: OutboxRow; error: string };
export type DrainSummary = { claimed: number; delivered: number; skipped: number; released: number; retried: number; dead: number; /** Time until the earliest backoff this drain wrote, so the caller can wake. */ nextRetryMs: number | null;};
let shuttingDown = false;
// Rows whose provider call has been dispatched and not yet resolved. Shutdown// must not hand these to the peer machine: the send can still succeed after the// row is released, and the peer would deliver it a second time.const inFlightSends = new Set<number>();const activeDrains = new Set<Promise<DrainSummary>>();
const SEND_METHOD = { alert: "sendAlert", recovery: "sendRecovery", degraded: "sendDegraded",} as const;
function backoffMs(attempt: number): number { const ceiling = Math.min(1000 * 2 ** attempt, MAX_BACKOFF_MS); return ceiling / 2 + Math.random() * (ceiling / 2);}
function errorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error);}
/** * Claims only the oldest pending row per (monitor, notification), so an alert * and the recovery behind it can never be delivered out of order, including * across machines. `rolloutPct` gates which monitors the drainer owns; at 0 the * inline sender is still authoritative and nothing here is claimable. */async function claimRows( limit: number, rolloutPct: number, lease: number, monitorIds?: number[],): Promise<OutboxRow[]> { const scope = monitorIds === undefined ? sql`` : sql` AND o.monitor_id IN (${sql.join( monitorIds.map((id) => sql`${id}`), sql`, `, )})`;
return withBusyRetry(() => db .update(notificationOutbox) .set({ lockedBy: workerId, lockedUntil: sql`min(unixepoch() + ${lease}, ${notificationOutbox.deadlineAt})`, attempts: sql`${notificationOutbox.attempts} + 1`, }) .where( sql`${notificationOutbox.id} IN ( SELECT o.id FROM ${notificationOutbox} o WHERE o.delivery_status = 'pending' AND (o.monitor_id % 100) < ${rolloutPct} AND o.next_attempt_at <= unixepoch() AND o.deadline_at > unixepoch()${scope} AND (o.locked_until IS NULL OR o.locked_until < unixepoch()) AND NOT EXISTS ( SELECT 1 FROM ${notificationOutbox} older WHERE older.monitor_id = o.monitor_id AND older.notification_id = o.notification_id AND older.delivery_status = 'pending' AND older.id < o.id) ORDER BY o.id LIMIT ${limit})`, ) .returning(), );}
type DeliveryDeps = { monitors: Map<number, Monitor>; notifications: Map<number, Notification>; incidents: Map<number, MonitorIncident>; smsBlocked: Map<number, boolean>;};
async function loadDeps(rows: OutboxRow[]): Promise<DeliveryDeps> { const monitorIds = [...new Set(rows.map((row) => row.monitorId))]; const notificationIds = [...new Set(rows.map((row) => row.notificationId))]; const incidentIds = [ ...new Set( rows .map((row) => row.incidentId) .filter((id): id is number => id !== null), ), ];
const [monitorRows, notificationRows, incidentRows] = await withBusyRetry( () => db.batch([ db.select().from(monitor).where(inArray(monitor.id, monitorIds)), db .select() .from(notification) .where(inArray(notification.id, notificationIds)), db .select() .from(monitorIncidentTable) .where( incidentIds.length === 0 ? sql`1 = 0` : inArray(monitorIncidentTable.id, incidentIds), ), ]), );
const monitors = new Map<number, Monitor>(); for (const row of monitorRows) { const parsed = selectMonitorSchema.safeParse(row); if (parsed.success) monitors.set(row.id, parsed.data); }
const notifications = new Map<number, Notification>(); for (const row of notificationRows) { const parsed = selectNotificationSchema.safeParse(row); if (parsed.success) notifications.set(row.id, parsed.data); }
const incidents = new Map<number, MonitorIncident>(); for (const row of incidentRows) { incidents.set(row.id, row); }
const smsWorkspaces = [ ...new Set( rows .filter((row) => row.provider === "sms") .map((row) => row.workspaceId) .filter((id): id is number => id !== null), ), ]; const smsBlocked = await loadSmsQuotaBlocked(smsWorkspaces);
return { monitors, notifications, incidents, smsBlocked };}
function deliverRow( row: OutboxRow, deps: DeliveryDeps, timeoutMs: number,): Effect.Effect<DeliveryOutcome> { return Effect.gen(function* () { if (shuttingDown) return { kind: "released", row } as const;
const monitorRow = deps.monitors.get(row.monitorId); const notificationRow = deps.notifications.get(row.notificationId); if (!monitorRow || !notificationRow) { return { kind: "dead", row, error: "monitor or notification no longer exists", } as const; }
if (row.provider === "sms" && row.workspaceId !== null) { if (deps.smsBlocked.get(row.workspaceId) === true) { return { kind: "skipped", row, reason: "sms-quota-exceeded" } as const; } }
const send = providerToFunction[row.provider][SEND_METHOD[row.eventType]]; const context = { monitor: monitorRow, notification: notificationRow, statusCode: row.payload.statusCode, message: row.payload.message, cronTimestamp: row.cronTimestamp, regions: row.payload.regions, latency: row.payload.latency, incident: row.incidentId === null ? undefined : deps.incidents.get(row.incidentId), };
inFlightSends.add(row.id); const exit = yield* Effect.exit( Effect.tryPromise({ try: () => send(context), catch: (error) => new Error(errorMessage(error)), }).pipe(Effect.timeout(timeoutMs)), ); inFlightSends.delete(row.id);
if (Exit.isSuccess(exit)) return { kind: "delivered", row } as const;
// The backoff is written to the row rather than slept through, so a retry // survives this process and either machine can pick it up. const error = String(exit.cause); const delayMs = backoffMs(row.attempts); if (Date.now() + delayMs >= row.deadlineAt * 1000) { return { kind: "dead", row, error } as const; }
return { kind: "retry", row, error, delayMs, nextAttemptAt: Math.ceil((Date.now() + delayMs) / 1000), } as const; });}
async function commitDelivered(rows: OutboxRow[]): Promise<void> { if (rows.length === 0) return; const ids = rows.map((row) => row.id);
await withBusyRetry(() => db.batch([ db .update(notificationOutbox) .set({ deliveryStatus: "settled", outcome: "delivered", deliveredAt: Math.floor(Date.now() / 1000), lockedBy: null, lockedUntil: null, }) .where(inArray(notificationOutbox.id, ids)), db .insert(notificationTrigger) .values( rows.map((row) => ({ monitorId: row.monitorId, notificationId: row.notificationId, cronTimestamp: row.cronTimestamp, })), ) .onConflictDoNothing(), ]), );
// Best-effort: the rows are settled and the triggers written. Throwing here // would abort the rest of the drain's commits and take the consumer loop down // with it, for a telemetry write. try { await Promise.all( rows.map((row) => checkerAudit.publishAuditLog({ id: `monitor:${row.monitorId}`, action: "notification.sent", targets: [{ id: String(row.monitorId), type: "monitor" }], metadata: { provider: row.provider, cronTimestamp: row.cronTimestamp, type: row.eventType, notificationId: row.notificationId, }, }), ), ); } catch (error) { logger.warn("Failed to publish notification audit log", { delivered_count: rows.length, error_message: errorMessage(error), }); }}
async function commitSkipped( entries: { row: OutboxRow; reason: string }[],): Promise<void> { const [first, ...rest] = entries.map((entry) => db .update(notificationOutbox) .set({ deliveryStatus: "settled", outcome: "skipped", lockedBy: null, lockedUntil: null, lastError: entry.reason, }) .where(eq(notificationOutbox.id, entry.row.id)), ); if (first === undefined) return; await withBusyRetry(() => db.batch([first, ...rest]));}
/** * Hands the row back with its backoff recorded. `attempts` already counts this * claim, so the row carries how many sends it has cost across every worker. */async function commitRetry( entries: { row: OutboxRow; error: string; nextAttemptAt: number }[],): Promise<void> { const [first, ...rest] = entries.map((entry) => db .update(notificationOutbox) .set({ nextAttemptAt: entry.nextAttemptAt, lockedBy: null, lockedUntil: null, lastError: entry.error.slice(0, 2000), }) .where(eq(notificationOutbox.id, entry.row.id)), ); if (first === undefined) return; await withBusyRetry(() => db.batch([first, ...rest]));}
async function commitReleased(rows: OutboxRow[]): Promise<void> { if (rows.length === 0) return; await withBusyRetry(() => db .update(notificationOutbox) .set({ lockedBy: null, lockedUntil: null }) .where( inArray( notificationOutbox.id, rows.map((row) => row.id), ), ), );}
async function commitDead( entries: { row: OutboxRow; error: string }[],): Promise<void> { if (entries.length === 0) return; const diedAt = Math.floor(Date.now() / 1000);
await withBusyRetry(() => db.batch([ db.insert(notificationDeadLetter).values( entries.map(({ row, error }) => ({ outboxId: row.id, dedupKey: row.dedupKey, monitorId: row.monitorId, workspaceId: row.workspaceId, notificationId: row.notificationId, provider: row.provider, eventType: row.eventType, fromStatus: row.fromStatus, toStatus: row.toStatus, cronTimestamp: row.cronTimestamp, incidentId: row.incidentId, payload: row.payload, attempts: row.attempts, finalError: error.slice(0, 2000), diedAt, })), ), db.delete(notificationOutbox).where( inArray( notificationOutbox.id, entries.map(({ row }) => row.id), ), ), ]), );
for (const { row, error } of entries) { logger.error("Notification dead-lettered", { monitor_id: row.monitorId, notification_id: row.notificationId, provider: row.provider, event_type: row.eventType, attempts: row.attempts, error_message: error, }); Sentry.captureException( new Error( `Notification dead-lettered: ${row.provider} for monitor ${row.monitorId}`, ), { tags: { provider: row.provider, event_type: row.eventType, from_status: row.fromStatus, to_status: row.toStatus, }, extra: { monitor_id: row.monitorId, notification_id: row.notificationId, workspace_id: row.workspaceId, attempts: row.attempts, final_error: error, }, }, ); }}
export type DrainOptions = { limit?: number; timeoutMs?: number; rolloutPct?: number; /** Restrict the drain to these monitors. */ monitorIds?: number[];};
/** One claim-and-deliver cycle. Safe to run concurrently on both machines. */export function drainOutbox(options: DrainOptions = {}): Promise<DrainSummary> { const drain = runDrain(options); activeDrains.add(drain); return drain.finally(() => { activeDrains.delete(drain); });}
async function runDrain(options: DrainOptions): Promise<DrainSummary> { const limit = options.limit ?? CLAIM_LIMIT; const timeoutMs = options.timeoutMs ?? env().NOTIFICATION_TIMEOUT_MS; const rolloutPct = options.rolloutPct ?? env().OUTBOX_ROLLOUT_PCT; const rows = await claimRows( limit, rolloutPct, leaseSeconds(limit, timeoutMs), options.monitorIds, ); if (rows.length === 0) { return { claimed: 0, delivered: 0, skipped: 0, released: 0, retried: 0, dead: 0, nextRetryMs: null, }; }
const deps = await loadDeps(rows);
const outcomes = await Effect.runPromise( Effect.forEach(rows, (row) => deliverRow(row, deps, timeoutMs), { concurrency: DELIVERY_CONCURRENCY, }), );
const delivered = outcomes .filter((outcome) => outcome.kind === "delivered") .map((outcome) => outcome.row); const skipped = outcomes.filter((outcome) => outcome.kind === "skipped"); const released = outcomes .filter((outcome) => outcome.kind === "released") .map((outcome) => outcome.row); const retried = outcomes.filter((outcome) => outcome.kind === "retry"); const dead = outcomes.filter((outcome) => outcome.kind === "dead");
await commitDelivered(delivered); await commitSkipped(skipped); await commitReleased(released); await commitRetry(retried); await commitDead(dead);
return { claimed: rows.length, delivered: delivered.length, skipped: skipped.length, released: released.length, retried: retried.length, dead: dead.length, nextRetryMs: retried.length === 0 ? null : Math.min(...retried.map((outcome) => outcome.delayMs)), };}
const EXPIRED_GRACE_SECONDS = 60 * 60;
/** * A row can expire without ever being claimed: the rollout gate excluded it, or * nothing drained for longer than the deadline. Claiming it later would page * someone about an outage that is long over, so the claim skips expired rows and * this sweep retires them. `attempts > 0` means a worker claimed it and it still * never landed, which is a dead letter; `attempts = 0` means we never owned it. */export async function sweepExpiredOutbox(): Promise<{ deadLettered: number; discarded: number;}> { const cutoff = Math.floor(Date.now() / 1000) - EXPIRED_GRACE_SECONDS;
const expired = await withBusyRetry(() => db .select() .from(notificationOutbox) .where( and( eq(notificationOutbox.deliveryStatus, "pending"), lt(notificationOutbox.deadlineAt, cutoff), ), ) .limit(200) .all(), );
if (expired.length === 0) return { deadLettered: 0, discarded: 0 };
const abandoned = expired.filter((row) => row.attempts > 0); const neverOwned = expired.filter((row) => row.attempts === 0);
await commitDead( abandoned.map((row) => ({ row, error: row.lastError ?? "expired before delivery completed", })), );
if (neverOwned.length > 0) { await withBusyRetry(() => db.delete(notificationOutbox).where( inArray( notificationOutbox.id, neverOwned.map((row) => row.id), ), ), ); logger.info("Discarded outbox rows that were never owned", { count: neverOwned.length, }); }
return { deadLettered: abandoned.length, discarded: neverOwned.length };}
/** Drains until the queue is empty, so one wake-up clears a whole burst. */export async function drainUntilEmpty( options: DrainOptions = {},): Promise<DrainSummary> { const total: DrainSummary = { claimed: 0, delivered: 0, skipped: 0, released: 0, retried: 0, dead: 0, nextRetryMs: null, };
const resolved: DrainOptions = { limit: options.limit ?? CLAIM_LIMIT, timeoutMs: options.timeoutMs ?? env().NOTIFICATION_TIMEOUT_MS, rolloutPct: options.rolloutPct ?? env().OUTBOX_ROLLOUT_PCT, monitorIds: options.monitorIds, };
while (!shuttingDown) { const summary = await drainOutbox(resolved); total.claimed += summary.claimed; total.delivered += summary.delivered; total.skipped += summary.skipped; total.released += summary.released; total.retried += summary.retried; total.dead += summary.dead; if (summary.nextRetryMs !== null) { total.nextRetryMs = total.nextRetryMs === null ? summary.nextRetryMs : Math.min(total.nextRetryMs, summary.nextRetryMs); } if (summary.claimed < (resolved.limit ?? CLAIM_LIMIT)) break; }
return total;}
const wakeQueue = Effect.runSync(Queue.make<number>());
/** The durable state is the outbox row; this only decides when we look. */export function enqueueOutbox(ids: number[]): void { for (const id of ids) { Queue.offerUnsafe(wakeQueue, id); }}
const retryTimers = new Set<ReturnType<typeof setTimeout>>();
/** * The backoff is already durable in `next_attempt_at` and the safety-net cron * would find it; this only makes the wait match the backoff instead of the cron. */function scheduleRetryWake(delayMs: number): void { if (shuttingDown) return; const timer = setTimeout(() => { retryTimers.delete(timer); Queue.offerUnsafe(wakeQueue, 0); }, delayMs); retryTimers.add(timer);}
export function startOutboxConsumer(): void { const loop = Effect.gen(function* () { while (!shuttingDown) { yield* Queue.takeAll(wakeQueue); const summary = yield* Effect.promise(() => drainUntilEmpty()); if (summary.nextRetryMs !== null) scheduleRetryWake(summary.nextRetryMs); } });
void Effect.runPromise(loop).catch((error) => { logger.error("Outbox consumer stopped", { error_message: errorMessage(error), }); });}
/** * Fly SIGTERMs on every deploy and on the daily restart. Releasing unstarted * claims hands them to the peer machine now instead of at lease expiry. */export async function shutdownOutbox(graceMs?: number): Promise<void> { shuttingDown = true; for (const timer of retryTimers) clearTimeout(timer); retryTimers.clear(); await Effect.runPromise(Queue.shutdown(wakeQueue)); await waitForDrains(graceMs ?? env().NOTIFICATION_TIMEOUT_MS + 5_000);
// Anything still sending keeps its lease: the provider call can succeed after // we stop watching, so the peer waits for the lease to lapse rather than // starting a second send while this one is outstanding. const stillSending = [...inFlightSends]; await withBusyRetry(() => db .update(notificationOutbox) .set({ lockedBy: null, lockedUntil: null }) .where( stillSending.length === 0 ? eq(notificationOutbox.lockedBy, workerId) : and( eq(notificationOutbox.lockedBy, workerId), notInArray(notificationOutbox.id, stillSending), ), ), );}
async function waitForDrains(graceMs: number): Promise<void> { if (activeDrains.size === 0) return; let timer: ReturnType<typeof setTimeout> | undefined; const grace = new Promise<void>((resolve) => { timer = setTimeout(resolve, graceMs); }); await Promise.race([ Promise.allSettled(activeDrains).then(() => undefined), grace, ]); if (timer !== undefined) clearTimeout(timer);}