perf(server): allow Neon to scale to zero when idle (#2217)

This commit is contained in:
RainbowBird
2026-08-04 10:58:27 +00:00
committed by GitHub
parent a239ebbb06
commit 04485788d0
12 changed files with 338 additions and 733 deletions
+5 -18
View File
@@ -49,11 +49,8 @@ import { resolveRequestAuth } from './libs/request-auth'
import { createUnauthorizedWsEvents } from './libs/ws-auth'
import { sessionMiddleware } from './middlewares/auth'
import { emitOtelLog, initOtel } from './otel'
import { registerActiveSessionsGauge } from './otel/gauges/active-sessions'
import { registerDistinctActiveUsersGauge } from './otel/gauges/distinct-active-users'
import { registerRollingActiveUsersGauge } from './otel/gauges/rolling-active-users'
import { registerTotalUsersGauge } from './otel/gauges/total-users'
import { registerTtsPoolGauge } from './otel/gauges/tts-pool'
import { createDiscardingUserMetricsSnapshotRecorder, registerUserMetricsSnapshotGauges } from './otel/gauges/user-metrics-snapshot'
import { registerWsOnlineUsersGauge } from './otel/gauges/ws-online-users'
import { createAdminRoutes } from './routes/admin'
import { createAdminUiRoutes } from './routes/admin-ui'
@@ -128,6 +125,9 @@ interface AppDeps {
export async function buildApp(deps: AppDeps) {
const logger = useLogger('app').useGlobalConfig()
const userMetricsRecorder = deps.otel
? registerUserMetricsSnapshotGauges(deps.otel.auth)
: createDiscardingUserMetricsSnapshotRecorder()
const app = new Hono<HonoEnv>()
.use('*', async (c, next) => {
@@ -456,6 +456,7 @@ export async function buildApp(deps: AppDeps) {
db: deps.db,
billingService: deps.billingService,
configKV: deps.configKV,
userMetricsRecorder,
}))
/**
@@ -823,21 +824,7 @@ export async function createApp() {
providerCatalogService,
ttsConcurrencyLedger,
})
// Register the cluster-wide ObservableGauges for sessions / users. Each
// replica polls the same DB (cached inside each gauge, in-flight coalesced);
// dashboards aggregate with avg()/max(), not sum(). See
// observability-conventions.md.
//
// Both gauges share the same `session` table: `user.active_sessions` is
// `COUNT(*)` (row inflation prone), `user.distinct_active` is
// `COUNT(DISTINCT user_id)` (real active-user count). Comparing the two
// surfaces session-row leakage from missing GC + per-OIDC-token row
// creation.
if (resolved.otel) {
registerTotalUsersGauge(resolved.otel.auth.totalUsers, resolved.db, resolved.otel.observability.metricReadErrors)
registerActiveSessionsGauge(resolved.otel.auth.activeSessions, resolved.db, resolved.otel.observability.metricReadErrors)
registerDistinctActiveUsersGauge(resolved.otel.auth.distinctActiveUsers, resolved.db, resolved.otel.observability.metricReadErrors)
registerRollingActiveUsersGauge(resolved.otel.auth.rollingActiveUsers, resolved.db, resolved.otel.observability.metricReadErrors)
registerTtsPoolGauge(resolved.otel.gateway.poolInflight, resolved.ttsConcurrencyLedger, resolved.otel.observability.metricReadErrors)
registerWsOnlineUsersGauge(resolved.otel.engagement.wsUsersOnline, resolved.redis, resolved.otel.observability.metricReadErrors)
}
@@ -1,102 +0,0 @@
import type { AuthMetrics, ObservabilityMetrics } from '..'
import type { Database } from '../../libs/db'
import { useLogger } from '@guiiai/logg'
import { count, gt } from 'drizzle-orm'
import { session as sessionTable } from '../../schemas/accounts'
/**
* Wire the `user.active_sessions` ObservableGauge to a Postgres `COUNT(*)`
* over the Better Auth session table.
*
* Use when:
* - Assembling DI in `createApp()`, exactly once per process.
*
* Expects:
* - `gauge` is the ObservableGauge handle created in `initOtel`.
* - `db` is the migrated Drizzle handle.
* - `metricReadErrors` is the shared counter used to track failures inside
* metric callbacks — increments are labelled with the originating metric
* name so on-call can spot which gauge is degraded.
*
* Multi-replica note:
* - This is a cluster-wide gauge — every replica reads the same DB and
* reports the same value. Dashboards MUST aggregate with `avg()`, NOT
* `sum()`. See observability-conventions.md.
*
* Concurrency:
* - Multiple OTel collection cycles can race (forced flushes, multiple
* readers). The in-flight promise lock keeps at most one DB query in
* flight per process; all other concurrent callbacks await the same
* result instead of stampeding the DB.
*
* Failure mode:
* - On DB error we increment `airi.observability.read_errors{metric}` and
* intentionally DO NOT call `result.observe(...)`. Letting the gauge
* skip an export cycle lets Prometheus staleness handle "DB is broken"
* correctly — an absence-based alert will fire after ~5 minutes. The
* previous version silently observed the stale cached value forever,
* which masked permanent DB failures.
*/
export function registerActiveSessionsGauge(
gauge: AuthMetrics['activeSessions'],
db: Database,
metricReadErrors: ObservabilityMetrics['metricReadErrors'],
) {
const log = useLogger('active-sessions-gauge').useGlobalConfig()
const CACHE_TTL_MS = 10_000
let cachedAt = 0
let cachedCount = 0
// Single shared promise representing "a refresh is in progress". All
// callbacks that arrive during a refresh attach to this and observe the
// same outcome. Reset to null when the refresh resolves.
let refreshInFlight: Promise<boolean> | null = null
async function refresh(): Promise<boolean> {
try {
// Use the app clock (`new Date()`) rather than DB clock (`NOW()`) so
// we agree with Better Auth's own session validity check, which uses
// `new Date()` in its session lookup (`better-auth/dist/session.mjs`
// and `dist/internal-adapter.mjs`). A DB/app clock skew would
// otherwise let this gauge disagree with auth-layer reality.
const rows = await db
.select({ count: count() })
.from(sessionTable)
.where(gt(sessionTable.expiresAt, new Date()))
cachedCount = Number(rows[0]?.count ?? 0)
cachedAt = Date.now()
return true
}
catch (err) {
log.withError(err).warn('Failed to read active sessions for gauge')
metricReadErrors.add(1, { metric: 'user.active_sessions' })
return false
}
}
gauge.addCallback(async (result) => {
const now = Date.now()
// Cache fresh — serve last good value without touching the DB.
if (cachedAt !== 0 && now - cachedAt < CACHE_TTL_MS) {
result.observe(cachedCount)
return
}
// Coalesce concurrent refreshes onto one in-flight promise.
if (!refreshInFlight) {
refreshInFlight = refresh().finally(() => {
refreshInFlight = null
})
}
const ok = await refreshInFlight
if (ok) {
result.observe(cachedCount)
}
// else: deliberately do nothing — let Prometheus staleness expose the
// outage instead of masking it with a stale cached number.
})
}
@@ -1,94 +0,0 @@
import type { AuthMetrics, ObservabilityMetrics } from '..'
import type { Database } from '../../libs/db'
import { useLogger } from '@guiiai/logg'
import { countDistinct, gt } from 'drizzle-orm'
import { session as sessionTable } from '../../schemas/accounts'
/**
* Wire the `user.distinct_active` ObservableGauge to a Postgres
* `COUNT(DISTINCT user_id)` over the Better Auth session table.
*
* Use when:
* - Assembling DI in `createApp()`, exactly once per process.
*
* Why this exists alongside `registerActiveSessionsGauge`:
* - `user.active_sessions` is `COUNT(*)` — counts session **rows**. Better
* Auth creates a new row per sign-in and per OIDC access-token issuance
* (the `oauth_access_token` table has a FK to `session.id`) and never GCs
* expired rows, so the row-count drifts up over time independently of
* the real user base. We've seen this metric show ~80K on a small
* deployment where the real distinct-user count is ~hundreds.
* - `user.distinct_active` is `COUNT(DISTINCT user_id)` — the actual
* active-user gauge. Pair with `user.active_sessions` to spot session
* inflation: if rows / users ratio climbs past ~5 it's probably time
* to add a session-GC cron or shorten Better Auth's `expiresIn`.
*
* Multi-replica note:
* - Cluster-wide gauge — every replica reads the same DB and reports the
* same value. Dashboards MUST aggregate with `avg()`, NOT `sum()`. See
* observability-conventions.md.
*
* Concurrency:
* - Same in-flight promise lock pattern as `registerActiveSessionsGauge`,
* so concurrent OTel collection cycles fold into one DB query.
*
* Failure mode:
* - DB error → increment `airi.observability.read_errors{metric}` and skip
* `result.observe(...)`. Prometheus staleness exposes the outage instead
* of pinning a stale cached value forever.
*/
export function registerDistinctActiveUsersGauge(
gauge: AuthMetrics['distinctActiveUsers'],
db: Database,
metricReadErrors: ObservabilityMetrics['metricReadErrors'],
) {
const log = useLogger('distinct-active-users-gauge').useGlobalConfig()
const CACHE_TTL_MS = 10_000
let cachedAt = 0
let cachedCount = 0
let refreshInFlight: Promise<boolean> | null = null
async function refresh(): Promise<boolean> {
try {
// Use the app clock (`new Date()`) for the same reason as
// `registerActiveSessionsGauge`: agree with Better Auth's own
// session-validity check, which uses `new Date()` rather than
// `NOW()`.
const rows = await db
.select({ count: countDistinct(sessionTable.userId) })
.from(sessionTable)
.where(gt(sessionTable.expiresAt, new Date()))
cachedCount = Number(rows[0]?.count ?? 0)
cachedAt = Date.now()
return true
}
catch (err) {
log.withError(err).warn('Failed to read distinct active users for gauge')
metricReadErrors.add(1, { metric: 'user.distinct_active' })
return false
}
}
gauge.addCallback(async (result) => {
const now = Date.now()
if (cachedAt !== 0 && now - cachedAt < CACHE_TTL_MS) {
result.observe(cachedCount)
return
}
if (!refreshInFlight) {
refreshInFlight = refresh().finally(() => {
refreshInFlight = null
})
}
const ok = await refreshInFlight
if (ok) {
result.observe(cachedCount)
}
})
}
@@ -1,138 +0,0 @@
import type { AuthMetrics, ObservabilityMetrics } from '..'
import type { Database } from '../../libs/db'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { registerRollingActiveUsersGauge } from './rolling-active-users'
/**
* Capture the callback registered via `gauge.addCallback` and a spyable
* `observe` so tests can drive collection cycles by hand.
*
* @example
* const { gauge, observe, run } = makeGauge()
* register...(gauge, db, errs)
* await run()
* expect(observe).toHaveBeenCalledWith(5, { window: '24h' })
*/
function makeGauge() {
let cb: ((result: { observe: (v: number, attrs: Record<string, string>) => void }) => void | Promise<void>) | null = null
const observe = vi.fn()
const gauge = {
addCallback: vi.fn((fn: typeof cb) => { cb = fn }),
} as unknown as AuthMetrics['rollingActiveUsers']
return {
gauge,
observe,
run: async () => {
if (!cb)
throw new Error('no callback registered')
await cb({ observe })
},
}
}
/**
* Drizzle's `db.select({...}).from(table)` is awaited directly (thenable
* query builder). Mock it as `select -> { from: () => Promise<rows> }`.
*
* @example
* const db = makeDb([{ dau: '5', wau: '10', mau: '20' }])
*/
function makeDb(rows: unknown[], opts: { reject?: boolean } = {}) {
const from = vi.fn(() => (opts.reject ? Promise.reject(new Error('db down')) : Promise.resolve(rows)))
const select = vi.fn(() => ({ from }))
return { db: { select } as unknown as Database, select, from }
}
function makeReadErrors() {
const add = vi.fn()
return { metricReadErrors: { add } as unknown as ObservabilityMetrics['metricReadErrors'], add }
}
describe('registerRollingActiveUsersGauge', () => {
beforeEach(() => {
vi.useFakeTimers()
})
afterEach(() => {
vi.useRealTimers()
vi.restoreAllMocks()
})
it('observes one point per window with the window attribute', async () => {
// Postgres count(*) comes back as a numeric string — assert we coerce it.
const { db } = makeDb([{ dau: '5', wau: '12', mau: '40' }])
const { metricReadErrors } = makeReadErrors()
const { gauge, observe, run } = makeGauge()
registerRollingActiveUsersGauge(gauge, db, metricReadErrors)
await run()
expect(observe).toHaveBeenCalledTimes(3)
expect(observe).toHaveBeenNthCalledWith(1, 5, { window: '24h' })
expect(observe).toHaveBeenNthCalledWith(2, 12, { window: '7d' })
expect(observe).toHaveBeenNthCalledWith(3, 40, { window: '30d' })
})
it('serves cached values without re-querying inside the 60s TTL', async () => {
const { db, select } = makeDb([{ dau: '3', wau: '7', mau: '9' }])
const { metricReadErrors } = makeReadErrors()
const { gauge, observe, run } = makeGauge()
registerRollingActiveUsersGauge(gauge, db, metricReadErrors)
await run()
await vi.advanceTimersByTimeAsync(30_000)
await run()
// Second collection within TTL hits cache: still one DB query, but six
// observes (3 per cycle).
expect(select).toHaveBeenCalledTimes(1)
expect(observe).toHaveBeenCalledTimes(6)
})
it('re-queries after the TTL expires', async () => {
const { db, select } = makeDb([{ dau: '1', wau: '1', mau: '1' }])
const { metricReadErrors } = makeReadErrors()
const { gauge, run } = makeGauge()
registerRollingActiveUsersGauge(gauge, db, metricReadErrors)
await run()
await vi.advanceTimersByTimeAsync(61_000)
await run()
expect(select).toHaveBeenCalledTimes(2)
})
it('on DB error: increments read-errors and does not observe', async () => {
// ROOT CAUSE:
//
// A silent gauge that observes a stale/zero value on DB failure masks the
// outage forever — an absence-based alert can never fire.
//
// We fixed this by skipping result.observe(...) on error and bumping
// airi.observability.read_errors{metric}, so Prometheus staleness exposes
// the broken DB instead.
const { db } = makeDb([], { reject: true })
const { metricReadErrors, add } = makeReadErrors()
const { gauge, observe, run } = makeGauge()
registerRollingActiveUsersGauge(gauge, db, metricReadErrors)
await run()
expect(observe).not.toHaveBeenCalled()
expect(add).toHaveBeenCalledWith(1, { metric: 'user.active_rolling' })
})
it('coalesces concurrent collection cycles into one DB query', async () => {
const { db, select } = makeDb([{ dau: '2', wau: '4', mau: '6' }])
const { metricReadErrors } = makeReadErrors()
const { gauge, run } = makeGauge()
registerRollingActiveUsersGauge(gauge, db, metricReadErrors)
// Fire two callbacks before the first refresh resolves — the in-flight
// lock must fold them into a single query.
await Promise.all([run(), run()])
expect(select).toHaveBeenCalledTimes(1)
})
})
@@ -1,132 +0,0 @@
import type { AuthMetrics, ObservabilityMetrics } from '..'
import type { Database } from '../../libs/db'
import { useLogger } from '@guiiai/logg'
import { sql } from 'drizzle-orm'
import { user as userTable } from '../../schemas/accounts'
/**
* Trailing windows reported by the `user.active_rolling` gauge. The `label`
* becomes the Prometheus `window` attribute (`user_active_rolling{window="24h"}`)
* and `ms` is the lookback applied to `user.last_seen_at`.
*/
const WINDOWS = [
{ label: '24h', ms: 24 * 60 * 60 * 1000 },
{ label: '7d', ms: 7 * 24 * 60 * 60 * 1000 },
{ label: '30d', ms: 30 * 24 * 60 * 60 * 1000 },
] as const
type WindowLabel = (typeof WINDOWS)[number]['label']
type RollingCounts = Record<WindowLabel, number>
/**
* Wire the `user.active_rolling` ObservableGauge to a Postgres
* `COUNT(*) FILTER (WHERE last_seen_at > cutoff)` over the `user` table, one
* filter per trailing window (DAU / WAU / MAU).
*
* Use when:
* - Assembling DI in `createApp()`, exactly once per process.
*
* Why this exists alongside `registerDistinctActiveUsersGauge`:
* - `user.distinct_active` counts users with a currently non-expired session
* ("signed in right now"). It cannot answer "how many users came back this
* week" because expired sessions drop out.
* - `user.active_rolling` reads `user.last_seen_at` — touched on sign-in and
* on every OIDC access-token refresh (~hourly) — so it measures activity
* over a trailing window regardless of session state. This is the standard
* DAU / WAU / MAU engagement funnel.
*
* Expects:
* - `gauge` is the ObservableGauge handle created in `initOtel`.
* - `db` is the migrated Drizzle handle.
* - `metricReadErrors` is the shared counter; failures inside the callback
* increment it labelled with the originating metric name.
*
* Multi-replica note:
* - Cluster-wide gauge — every replica reads the same DB and reports the same
* value per window. Dashboards MUST aggregate with `max()`/`avg()`, NOT
* `sum()`. See observability-conventions.md.
*
* Concurrency:
* - Same in-flight promise lock as the sibling gauges, so concurrent OTel
* collection cycles fold into one DB query. Cached for 60s — DAU/WAU/MAU
* move slowly and the query scans the whole `user` table, so a longer TTL
* than the per-row session gauges keeps DB load low.
*
* Failure mode:
* - DB error → increment `airi.observability.read_errors{metric}` and skip
* `result.observe(...)`. Prometheus staleness exposes the outage instead of
* pinning stale values forever.
*/
export function registerRollingActiveUsersGauge(
gauge: AuthMetrics['rollingActiveUsers'],
db: Database,
metricReadErrors: ObservabilityMetrics['metricReadErrors'],
) {
const log = useLogger('rolling-active-users-gauge').useGlobalConfig()
const CACHE_TTL_MS = 60_000
let cachedAt = 0
let cached: RollingCounts = { '24h': 0, '7d': 0, '30d': 0 }
let refreshInFlight: Promise<boolean> | null = null
async function refresh(): Promise<boolean> {
try {
// Compute cutoffs from the app clock so all three windows are anchored
// to the same instant within one query. `id` is the PK, so
// `count(*)` == `count(distinct id)` — no DISTINCT needed.
const now = Date.now()
const since24h = new Date(now - WINDOWS[0].ms)
const since7d = new Date(now - WINDOWS[1].ms)
const since30d = new Date(now - WINDOWS[2].ms)
const rows = await db
.select({
dau: sql<number>`count(*) filter (where ${userTable.lastSeenAt} > ${since24h})`,
wau: sql<number>`count(*) filter (where ${userTable.lastSeenAt} > ${since7d})`,
mau: sql<number>`count(*) filter (where ${userTable.lastSeenAt} > ${since30d})`,
})
.from(userTable)
cached = {
'24h': Number(rows[0]?.dau ?? 0),
'7d': Number(rows[0]?.wau ?? 0),
'30d': Number(rows[0]?.mau ?? 0),
}
cachedAt = Date.now()
return true
}
catch (err) {
log.withError(err).warn('Failed to read rolling active users for gauge')
metricReadErrors.add(1, { metric: 'user.active_rolling' })
return false
}
}
function observeAll(result: Parameters<Parameters<typeof gauge.addCallback>[0]>[0]) {
for (const w of WINDOWS)
result.observe(cached[w.label], { window: w.label })
}
gauge.addCallback(async (result) => {
const now = Date.now()
// Cache fresh — serve last good values without touching the DB.
if (cachedAt !== 0 && now - cachedAt < CACHE_TTL_MS) {
observeAll(result)
return
}
// Coalesce concurrent refreshes onto one in-flight promise.
if (!refreshInFlight) {
refreshInFlight = refresh().finally(() => {
refreshInFlight = null
})
}
const ok = await refreshInFlight
if (ok)
observeAll(result)
// else: deliberately skip observe — let Prometheus staleness expose the
// DB outage instead of masking it with stale numbers.
})
}
@@ -1,106 +0,0 @@
import type { AuthMetrics, ObservabilityMetrics } from '..'
import type { Database } from '../../libs/db'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { registerTotalUsersGauge } from './total-users'
/**
* Capture the callback registered via `gauge.addCallback` and a spyable
* `observe` so tests can drive collection cycles by hand.
*
* @example
* const { gauge, observe, run } = makeGauge()
* registerTotalUsersGauge(gauge, db, errs)
* await run()
* expect(observe).toHaveBeenCalledWith(42)
*/
function makeGauge() {
let cb: ((result: { observe: (v: number) => void }) => void | Promise<void>) | null = null
const observe = vi.fn()
const gauge = {
addCallback: vi.fn((fn: typeof cb) => { cb = fn }),
} as unknown as AuthMetrics['totalUsers']
return {
gauge,
observe,
run: async () => {
if (!cb)
throw new Error('no callback registered')
await cb({ observe })
},
}
}
/**
* Drizzle's `db.select({...}).from(table)` is awaited directly (thenable
* query builder). Mock it as `select -> { from: () => Promise<rows> }`.
*
* @example
* const db = makeDb([{ count: '12' }])
*/
function makeDb(rows: unknown[], opts: { reject?: boolean } = {}) {
const from = vi.fn(() => (opts.reject ? Promise.reject(new Error('db down')) : Promise.resolve(rows)))
const select = vi.fn(() => ({ from }))
return { db: { select } as unknown as Database, select, from }
}
function makeReadErrors() {
const add = vi.fn()
return { metricReadErrors: { add } as unknown as ObservabilityMetrics['metricReadErrors'], add }
}
describe('registerTotalUsersGauge', () => {
beforeEach(() => {
vi.useFakeTimers()
})
afterEach(() => {
vi.useRealTimers()
vi.restoreAllMocks()
})
it('observes the total user count from the user table', async () => {
const { db } = makeDb([{ count: '42' }])
const { metricReadErrors } = makeReadErrors()
const { gauge, observe, run } = makeGauge()
registerTotalUsersGauge(gauge, db, metricReadErrors)
await run()
expect(observe).toHaveBeenCalledWith(42)
})
it('serves cached values without re-querying inside the 60s TTL', async () => {
const { db, select } = makeDb([{ count: '3' }])
const { metricReadErrors } = makeReadErrors()
const { gauge, observe, run } = makeGauge()
registerTotalUsersGauge(gauge, db, metricReadErrors)
await run()
await vi.advanceTimersByTimeAsync(30_000)
await run()
expect(select).toHaveBeenCalledTimes(1)
expect(observe).toHaveBeenCalledTimes(2)
})
it('on DB error: increments read-errors and does not observe', async () => {
// ROOT CAUSE:
//
// A total-user gauge that observes a stale/zero value on DB failure masks
// the outage forever — an absence-based alert can never fire.
//
// We fixed this by skipping result.observe(...) on error and bumping
// airi.observability.read_errors{metric}, so Prometheus staleness exposes
// the broken DB instead.
const { db } = makeDb([], { reject: true })
const { metricReadErrors, add } = makeReadErrors()
const { gauge, observe, run } = makeGauge()
registerTotalUsersGauge(gauge, db, metricReadErrors)
await run()
expect(observe).not.toHaveBeenCalled()
expect(add).toHaveBeenCalledWith(1, { metric: 'user.total' })
})
})
@@ -1,81 +0,0 @@
import type { AuthMetrics, ObservabilityMetrics } from '..'
import type { Database } from '../../libs/db'
import { useLogger } from '@guiiai/logg'
import { count } from 'drizzle-orm'
import { user as userTable } from '../../schemas/accounts'
/**
* Wire the `user.total` ObservableGauge to a Postgres `COUNT(*)` over the
* Better Auth user table.
*
* Use when:
* - Assembling DI in `createApp()`, exactly once per process.
*
* Expects:
* - `gauge` is the ObservableGauge handle created in `initOtel`.
* - `db` is the migrated Drizzle handle.
* - `metricReadErrors` is the shared counter used to track failures inside
* metric callbacks.
*
* Multi-replica note:
* - Cluster-wide gauge — every replica reads the same DB and reports the same
* value. Dashboards MUST aggregate with `max()`/`avg()`, NOT `sum()`.
*
* Concurrency:
* - Same in-flight promise lock as the sibling DB-backed gauges, so concurrent
* OTel collection cycles fold into one DB query.
*
* Failure mode:
* - DB error → increment `airi.observability.read_errors{metric}` and skip
* `result.observe(...)`. Prometheus staleness exposes the outage instead of
* pinning stale values forever.
*/
export function registerTotalUsersGauge(
gauge: AuthMetrics['totalUsers'],
db: Database,
metricReadErrors: ObservabilityMetrics['metricReadErrors'],
) {
const log = useLogger('total-users-gauge').useGlobalConfig()
const CACHE_TTL_MS = 60_000
let cachedAt = 0
let cachedCount = 0
let refreshInFlight: Promise<boolean> | null = null
async function refresh(): Promise<boolean> {
try {
const rows = await db
.select({ count: count() })
.from(userTable)
cachedCount = Number(rows[0]?.count ?? 0)
cachedAt = Date.now()
return true
}
catch (err) {
log.withError(err).warn('Failed to read total users for gauge')
metricReadErrors.add(1, { metric: 'user.total' })
return false
}
}
gauge.addCallback(async (result) => {
const now = Date.now()
if (cachedAt !== 0 && now - cachedAt < CACHE_TTL_MS) {
result.observe(cachedCount)
return
}
if (!refreshInFlight) {
refreshInFlight = refresh().finally(() => {
refreshInFlight = null
})
}
const ok = await refreshInFlight
if (ok)
result.observe(cachedCount)
})
}
@@ -0,0 +1,127 @@
import type { ObservableCallback, ObservableResult } from '@opentelemetry/api'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { registerUserMetricsSnapshotGauges, USER_METRICS_SNAPSHOT_MAX_AGE_MS } from './user-metrics-snapshot'
afterEach(() => {
vi.restoreAllMocks()
})
function createGaugeProbe() {
let callback: ObservableCallback | undefined
const observe = vi.fn()
return {
gauge: {
addCallback(value: ObservableCallback) {
callback = value
},
},
observe,
async collect() {
if (!callback)
throw new Error('Gauge callback was not registered')
const result: ObservableResult = { observe }
await callback(result)
},
}
}
describe('registerUserMetricsSnapshotGauges', () => {
it('keeps periodic collection passive and only observes an explicitly recorded snapshot', async () => {
const refreshedAt = Date.parse('2026-08-04T00:00:00.000Z')
vi.spyOn(Date, 'now').mockReturnValue(refreshedAt)
const totalUsers = createGaugeProbe()
const activeSessions = createGaugeProbe()
const distinctActiveUsers = createGaugeProbe()
const rollingActiveUsers = createGaugeProbe()
const recorder = registerUserMetricsSnapshotGauges({
totalUsers: totalUsers.gauge,
activeSessions: activeSessions.gauge,
distinctActiveUsers: distinctActiveUsers.gauge,
rollingActiveUsers: rollingActiveUsers.gauge,
})
await Promise.all([
totalUsers.collect(),
activeSessions.collect(),
distinctActiveUsers.collect(),
rollingActiveUsers.collect(),
])
expect(totalUsers.observe).not.toHaveBeenCalled()
expect(activeSessions.observe).not.toHaveBeenCalled()
expect(distinctActiveUsers.observe).not.toHaveBeenCalled()
expect(rollingActiveUsers.observe).not.toHaveBeenCalled()
recorder.record(
{
totalUsers: 42,
activeSessions: 7,
distinctActiveUsers: 5,
rollingActiveUsers: {
'24h': 9,
'7d': 18,
'30d': 30,
},
},
refreshedAt,
)
await Promise.all([
totalUsers.collect(),
activeSessions.collect(),
distinctActiveUsers.collect(),
rollingActiveUsers.collect(),
])
expect(totalUsers.observe).toHaveBeenLastCalledWith(42)
expect(activeSessions.observe).toHaveBeenLastCalledWith(7)
expect(distinctActiveUsers.observe).toHaveBeenLastCalledWith(5)
expect(rollingActiveUsers.observe).toHaveBeenNthCalledWith(1, 9, { window: '24h' })
expect(rollingActiveUsers.observe).toHaveBeenNthCalledWith(2, 18, { window: '7d' })
expect(rollingActiveUsers.observe).toHaveBeenNthCalledWith(3, 30, { window: '30d' })
})
it('stops observing a snapshot after its bounded freshness interval', async () => {
const refreshedAt = Date.parse('2026-08-04T00:00:00.000Z')
vi.spyOn(Date, 'now').mockReturnValue(refreshedAt)
const totalUsers = createGaugeProbe()
const activeSessions = createGaugeProbe()
const distinctActiveUsers = createGaugeProbe()
const rollingActiveUsers = createGaugeProbe()
const recorder = registerUserMetricsSnapshotGauges({
totalUsers: totalUsers.gauge,
activeSessions: activeSessions.gauge,
distinctActiveUsers: distinctActiveUsers.gauge,
rollingActiveUsers: rollingActiveUsers.gauge,
})
recorder.record(
{
totalUsers: 42,
activeSessions: 7,
distinctActiveUsers: 5,
rollingActiveUsers: { '24h': 9, '7d': 18, '30d': 30 },
},
refreshedAt,
)
vi.mocked(Date.now).mockReturnValue(refreshedAt + USER_METRICS_SNAPSHOT_MAX_AGE_MS)
await Promise.all([
totalUsers.collect(),
activeSessions.collect(),
distinctActiveUsers.collect(),
rollingActiveUsers.collect(),
])
expect(totalUsers.observe).not.toHaveBeenCalled()
expect(activeSessions.observe).not.toHaveBeenCalled()
expect(distinctActiveUsers.observe).not.toHaveBeenCalled()
expect(rollingActiveUsers.observe).not.toHaveBeenCalled()
})
})
@@ -0,0 +1,91 @@
import type { ObservableGauge } from '@opentelemetry/api'
const ROLLING_WINDOWS = ['24h', '7d', '30d'] as const
// The admin endpoint caches DB aggregates for 60 seconds. Two minutes keeps a
// continuously refreshed dashboard stable across cache rollover and several
// 15-second OTel collections, while still bounding stale replica dominance.
export const USER_METRICS_SNAPSHOT_MAX_AGE_MS = 2 * 60_000
type RollingWindow = (typeof ROLLING_WINDOWS)[number]
export interface UserMetricsSnapshot {
totalUsers: number
activeSessions: number
distinctActiveUsers: number
rollingActiveUsers: Record<RollingWindow, number>
}
export interface UserMetricsSnapshotRecorder {
record: (snapshot: UserMetricsSnapshot, refreshedAt: number) => void
}
type ObservableGaugeRegistration = Pick<ObservableGauge, 'addCallback'>
export interface UserMetricsSnapshotGauges {
totalUsers: ObservableGaugeRegistration
activeSessions: ObservableGaugeRegistration
distinctActiveUsers: ObservableGaugeRegistration
rollingActiveUsers: ObservableGaugeRegistration
}
/**
* Export the latest explicitly refreshed user-metrics snapshot without doing
* I/O from OTel's periodic collection callbacks. Until a request records the
* first snapshot, or after the last database refresh becomes stale, the
* gauges intentionally emit no points.
*/
export function registerUserMetricsSnapshotGauges(
gauges: UserMetricsSnapshotGauges,
): UserMetricsSnapshotRecorder {
let latest: { snapshot: UserMetricsSnapshot, refreshedAt: number } | undefined
function readFreshSnapshot(): UserMetricsSnapshot | undefined {
if (!latest)
return undefined
if (Date.now() - latest.refreshedAt >= USER_METRICS_SNAPSHOT_MAX_AGE_MS) {
latest = undefined
return undefined
}
return latest.snapshot
}
gauges.totalUsers.addCallback((result) => {
const snapshot = readFreshSnapshot()
if (snapshot)
result.observe(snapshot.totalUsers)
})
gauges.activeSessions.addCallback((result) => {
const snapshot = readFreshSnapshot()
if (snapshot)
result.observe(snapshot.activeSessions)
})
gauges.distinctActiveUsers.addCallback((result) => {
const snapshot = readFreshSnapshot()
if (snapshot)
result.observe(snapshot.distinctActiveUsers)
})
gauges.rollingActiveUsers.addCallback((result) => {
const snapshot = readFreshSnapshot()
if (!snapshot)
return
for (const window of ROLLING_WINDOWS)
result.observe(snapshot.rollingActiveUsers[window], { window })
})
return {
record(snapshot, refreshedAt) {
latest = { snapshot, refreshedAt }
},
}
}
export function createDiscardingUserMetricsSnapshotRecorder(): UserMetricsSnapshotRecorder {
return { record() {} }
}
+30 -27
View File
@@ -76,39 +76,42 @@ export interface AuthMetrics {
userRegistered: Counter
userLogin: Counter
/**
* Pull-based gauge for total registered users.
* Gauge for the latest requested total registered-user snapshot.
*
* Use when:
* - Reporting current account-base size. Pair with
* {@link AuthMetrics.userRegistered} for signup deltas over a time window.
*
* Expects:
* - Backed by `SELECT COUNT(*) FROM "user"`. Same cluster-wide truth as the
* other DB-backed gauges; dashboards MUST aggregate with `max()`/`avg()`,
* not `sum()`.
* - Refreshed when an authorized caller requests `/api/admin/metrics`.
* OTel collection only reads the in-process snapshot and never queries
* Postgres. The gauge stops emitting after the bounded freshness window
* so an old replica can become stale. Dashboards MUST aggregate with
* `max()`/`avg()`, not `sum()`.
*/
totalUsers: ObservableGauge
/**
* Cluster-wide active session count, sourced from Postgres (Better Auth
* `session` table where `expires_at > NOW()`).
* Latest requested active-session count from the Better Auth `session`
* table where `expires_at` is in the future.
*
* Why ObservableGauge instead of UpDownCounter:
* - UpDownCounter drifts: TTL expiration never fires a -1, and multi-
* replica deploys split +1 / -1 across instances (signin on A, signout
* on B). The previous implementation went unboundedly positive.
* - Reading from the source-of-truth DB at scrape time makes the metric
* self-correcting.
* - The admin metrics request refreshes this source-of-truth snapshot;
* periodic OTel collection does no I/O, so telemetry cannot keep a
* serverless database awake.
*
* Multi-replica note:
* - Every replica reads the same DB and reports the same value, so the
* - Only replicas with a fresh admin metrics snapshot emit a point. The
* dashboard MUST aggregate with `max()` (or `avg()`), NOT `sum()`.
* Using sum() would multiply the real count by the replica count.
* - See `server/apps/api/docs/ai-context/observability-conventions.md`,
* "Multi-Replica Considerations".
*/
activeSessions: ObservableGauge
/**
* Pull-based gauge for distinct users with ≥1 non-expired session.
* Gauge for the latest requested count of distinct users with at least one
* non-expired session.
*
* Use when:
* - Querying real "active users" — not session rows. Better Auth creates a
@@ -117,13 +120,13 @@ export interface AuthMetrics {
* over time even when the actual user base is small.
*
* Expects:
* - Backed by `SELECT COUNT(DISTINCT user_id) FROM session WHERE expires_at > now()`.
* Same cluster-wide truth as `activeSessions`; dashboards MUST aggregate
* with `avg()`, not `sum()` — see observability-conventions.md.
* - Refreshed by `/api/admin/metrics` using `COUNT(DISTINCT user_id)`.
* Periodic collection reads memory only; dashboards MUST aggregate with
* `avg()`/`max()`, not `sum()`.
*/
distinctActiveUsers: ObservableGauge
/**
* Pull-based gauge for rolling-window distinct active users (DAU / WAU /
* Gauge for the latest requested rolling-window active users (DAU / WAU /
* MAU).
*
* Use when:
@@ -133,13 +136,13 @@ export interface AuthMetrics {
* currently-live session.
*
* Expects:
* - Backed by `COUNT(*) FILTER (WHERE last_seen_at > now() - window)` over
* the `user` table. `last_seen_at` is touched on sign-in and on every
* - Refreshed by `/api/admin/metrics` using `COUNT(*) FILTER` over the
* `user` table. `last_seen_at` is touched on sign-in and on every
* OIDC access-token refresh (~hourly), so it is a per-user last-activity
* timestamp (see the `user.lastSeenAt` schema note).
* - Observed once per window with a `window` attribute (`24h` / `7d` /
* `30d`). Same cluster-wide truth as the other DB-backed gauges;
* dashboards MUST aggregate with `max()`/`avg()`, not `sum()`.
* `30d`). Periodic collection reads memory only; dashboards MUST
* aggregate with `max()`/`avg()`, not `sum()`.
*/
rollingActiveUsers: ObservableGauge
}
@@ -325,10 +328,10 @@ export interface RateLimitMetrics {
export interface ObservabilityMetrics {
/**
* Counts failures inside metric-pipeline callbacks (e.g. a DB-backed
* ObservableGauge that couldn't read from Postgres). Use for self-monitoring
* when this is rising, treat the affected gauge's reported value as
* potentially stale.
* Counts failures inside metric-pipeline callbacks (for example, a Redis-
* backed ObservableGauge that could not refresh). Use for self-monitoring
* when this is rising, treat the affected gauge's value as potentially
* stale.
*
* Labels: `metric` (the failing gauge's logical name).
*/
@@ -402,16 +405,16 @@ export function initOtel(env: Env): OtelInstance | null {
description: 'Number of user sign-ins',
}),
totalUsers: meter.createObservableGauge(METRIC_USER_TOTAL, {
description: 'Total registered users sourced from Postgres (cluster-wide; dashboard must use max(), not sum())',
description: 'Fresh admin-requested total registered-user snapshot (memory-only and expires when stale; dashboard must use max(), not sum())',
}),
activeSessions: meter.createObservableGauge(METRIC_USER_ACTIVE_SESSIONS, {
description: 'Active user sessions sourced from Postgres (cluster-wide; dashboard must use avg(), not sum())',
description: 'Fresh admin-requested active-session snapshot (memory-only and expires when stale; dashboard must use max(), not sum())',
}),
distinctActiveUsers: meter.createObservableGauge(METRIC_USER_DISTINCT_ACTIVE, {
description: 'Distinct users with ≥1 non-expired session — true active-user count, immune to per-row session inflation (cluster-wide; dashboard must use avg(), not sum())',
description: 'Fresh admin-requested distinct active-user snapshot, immune to per-row session inflation and omitted when stale (dashboard must use max(), not sum())',
}),
rollingActiveUsers: meter.createObservableGauge(METRIC_USER_ACTIVE_ROLLING, {
description: 'Rolling-window distinct active users (DAU/WAU/MAU) from user.last_seen_at, labelled by window=24h|7d|30d (cluster-wide; dashboard must use max(), not sum())',
description: 'Fresh admin-requested DAU/WAU/MAU snapshot from user.last_seen_at, omitted when stale and labelled by window=24h|7d|30d (dashboard must use max(), not sum())',
}),
}
+21 -2
View File
@@ -18,6 +18,7 @@ describe('admin metrics', () => {
vi.spyOn(Date, 'now').mockImplementation(() => now)
const db = await mockDB(schema)
const recordUserMetrics = vi.fn()
await db.insert(schema.user).values({
id: 'admin-1',
name: 'Admin',
@@ -48,11 +49,24 @@ describe('admin metrics', () => {
db,
billingService: {} as never,
configKV: {} as never,
userMetricsRecorder: { record: recordUserMetrics },
}))
const firstResponse = await app.request('/api/admin/metrics')
expect(firstResponse.status).toBe(200)
expect(await firstResponse.json()).toMatchObject({ totalUsers: 1, verifiedUsers: 1, adminSeats: 1 })
expect(await firstResponse.json()).toMatchObject({
totalUsers: 1,
verifiedUsers: 1,
adminSeats: 1,
activeSessions: 0,
distinctActiveUsers: 0,
rollingActiveUsers: { '24h': 0, '7d': 0, '30d': 0 },
})
expect(recordUserMetrics).toHaveBeenLastCalledWith(expect.objectContaining({
totalUsers: 1,
activeSessions: 0,
distinctActiveUsers: 0,
}), now)
await db.insert(schema.user).values({
id: 'user-2',
@@ -61,13 +75,18 @@ describe('admin metrics', () => {
emailVerified: false,
})
now += 30_000
const cachedResponse = await app.request('/api/admin/metrics')
expect(cachedResponse.status).toBe(200)
expect(await cachedResponse.json()).toMatchObject({ totalUsers: 1, verifiedUsers: 1, adminSeats: 1 })
expect(recordUserMetrics).toHaveBeenCalledTimes(2)
expect(recordUserMetrics).toHaveBeenLastCalledWith(expect.anything(), Date.parse('2026-08-04T00:00:00.000Z'))
now += 60_001
now += 30_001
const refreshedResponse = await app.request('/api/admin/metrics')
expect(refreshedResponse.status).toBe(200)
expect(await refreshedResponse.json()).toMatchObject({ totalUsers: 2, verifiedUsers: 1, adminSeats: 1 })
expect(recordUserMetrics).toHaveBeenCalledTimes(3)
expect(recordUserMetrics).toHaveBeenLastCalledWith(expect.anything(), now)
})
})
+64 -33
View File
@@ -1,11 +1,12 @@
import type { Context } from 'hono'
import type { Database } from '../../libs/db'
import type { UserMetricsSnapshot, UserMetricsSnapshotRecorder } from '../../otel/gauges/user-metrics-snapshot'
import type { ConfigKVService } from '../../services/adapters/config-kv'
import type { BillingService } from '../../services/domain/billing/billing-service'
import type { HonoEnv } from '../../types/hono'
import { and, asc, count, desc, eq, gt, ilike, isNull, or, sql } from 'drizzle-orm'
import { and, asc, count, countDistinct, desc, eq, gt, ilike, isNull, or, sql } from 'drizzle-orm'
import { Hono } from 'hono'
import { integer, maxLength, maxValue, minValue, nonEmpty, number, object, optional, pipe, safeParse, string } from 'valibot'
@@ -57,12 +58,11 @@ export interface AdminRoutesDeps {
db: Database
billingService: BillingService
configKV: ConfigKVService
userMetricsRecorder: UserMetricsSnapshotRecorder
}
interface AdminMetricsSnapshot {
totalUsers: number
interface AdminMetricsSnapshot extends UserMetricsSnapshot {
verifiedUsers: number
activeSessions: number
currentFlux: number
issuedFlux: number
llmRequests24h: number
@@ -71,14 +71,19 @@ interface AdminMetricsSnapshot {
grafanaEmbedUrl: null
}
function createAdminMetricsReader(db: Database) {
let cached: { value: AdminMetricsSnapshot, expiresAt: number } | undefined
let inFlight: Promise<AdminMetricsSnapshot> | undefined
interface AdminMetricsRead {
value: AdminMetricsSnapshot
refreshedAt: number
}
return async function readAdminMetrics(): Promise<AdminMetricsSnapshot> {
function createAdminMetricsReader(db: Database) {
let cached: (AdminMetricsRead & { expiresAt: number }) | undefined
let inFlight: Promise<AdminMetricsRead> | undefined
return async function readAdminMetrics(): Promise<AdminMetricsRead> {
const now = Date.now()
if (cached && cached.expiresAt > now)
return cached.value
return cached
// A polling burst can arrive immediately after expiry. Share that refresh
// within this API process so only one set of aggregate queries reaches DB.
@@ -86,47 +91,71 @@ function createAdminMetricsReader(db: Database) {
return inFlight
inFlight = (async () => {
const yesterday = new Date(Date.now() - 24 * 60 * 60 * 1000)
const currentTime = Date.now()
const yesterday = new Date(currentTime - 24 * 60 * 60 * 1000)
const weekAgo = new Date(currentTime - 7 * 24 * 60 * 60 * 1000)
const monthAgo = new Date(currentTime - 30 * 24 * 60 * 60 * 1000)
const [
totalUsers,
verifiedUsers,
activeSessions,
users,
sessions,
currentFlux,
issuedFlux,
llmRequests24h,
llmFlux24h,
adminUsers,
llmUsage24h,
] = await Promise.all([
db.select({ count: count() }).from(userTable),
db.select({ count: count() }).from(userTable).where(eq(userTable.emailVerified, true)),
db.select({ count: count() }).from(sessionTable).where(gt(sessionTable.expiresAt, new Date())),
db.select({
totalUsers: count(),
verifiedUsers: sql<number>`count(*) filter (where ${userTable.emailVerified} = true)`,
adminSeats: sql<number>`count(*) filter (where 'admin' = any(regexp_split_to_array(coalesce(${userTable.role}, ''), '\\s*,\\s*')))`,
active24h: sql<number>`count(*) filter (where ${userTable.lastSeenAt} > ${yesterday})`,
active7d: sql<number>`count(*) filter (where ${userTable.lastSeenAt} > ${weekAgo})`,
active30d: sql<number>`count(*) filter (where ${userTable.lastSeenAt} > ${monthAgo})`,
}).from(userTable),
db
.select({
activeSessions: count(),
distinctActiveUsers: countDistinct(sessionTable.userId),
})
.from(sessionTable)
.where(gt(sessionTable.expiresAt, new Date(currentTime))),
db.select({ total: sql<number>`coalesce(sum(${userFlux.flux}), 0)::int` }).from(userFlux).where(isNull(userFlux.deletedAt)),
db.select({ total: sql<number>`coalesce(sum(${fluxTransaction.amount}) filter (where ${fluxTransaction.type} in ('credit', 'initial', 'promo')), 0)::int` }).from(fluxTransaction),
db.select({ count: count() }).from(llmRequestLog).where(gt(llmRequestLog.createdAt, yesterday)),
db.select({ total: sql<number>`coalesce(sum(${llmRequestLog.fluxConsumed}), 0)::int` }).from(llmRequestLog).where(gt(llmRequestLog.createdAt, yesterday)),
db
.select({ count: count() })
.from(userTable)
.where(sql<boolean>`'admin' = any(regexp_split_to_array(coalesce(${userTable.role}, ''), '\\s*,\\s*'))`),
.select({
count: count(),
total: sql<number>`coalesce(sum(${llmRequestLog.fluxConsumed}), 0)::int`,
})
.from(llmRequestLog)
.where(gt(llmRequestLog.createdAt, yesterday)),
])
const value: AdminMetricsSnapshot = {
totalUsers: Number(totalUsers[0]?.count ?? 0),
verifiedUsers: Number(verifiedUsers[0]?.count ?? 0),
activeSessions: Number(activeSessions[0]?.count ?? 0),
totalUsers: Number(users[0]?.totalUsers ?? 0),
verifiedUsers: Number(users[0]?.verifiedUsers ?? 0),
activeSessions: Number(sessions[0]?.activeSessions ?? 0),
distinctActiveUsers: Number(sessions[0]?.distinctActiveUsers ?? 0),
rollingActiveUsers: {
'24h': Number(users[0]?.active24h ?? 0),
'7d': Number(users[0]?.active7d ?? 0),
'30d': Number(users[0]?.active30d ?? 0),
},
currentFlux: Number(currentFlux[0]?.total ?? 0),
issuedFlux: Number(issuedFlux[0]?.total ?? 0),
llmRequests24h: Number(llmRequests24h[0]?.count ?? 0),
llmFlux24h: Number(llmFlux24h[0]?.total ?? 0),
adminSeats: Number(adminUsers[0]?.count ?? 0),
llmRequests24h: Number(llmUsage24h[0]?.count ?? 0),
llmFlux24h: Number(llmUsage24h[0]?.total ?? 0),
adminSeats: Number(users[0]?.adminSeats ?? 0),
grafanaEmbedUrl: null,
}
// Expiry starts after the refresh finishes; slow aggregate queries should
// not shorten the period during which the completed snapshot is reused.
cached = { value, expiresAt: Date.now() + ADMIN_METRICS_CACHE_TTL_MS }
return value
const refreshedAt = Date.now()
cached = {
value,
refreshedAt,
expiresAt: refreshedAt + ADMIN_METRICS_CACHE_TTL_MS,
}
return cached
})()
try {
@@ -244,7 +273,9 @@ export function createAdminRoutes(deps: AdminRoutesDeps) {
})
.get('/metrics', async (c) => {
return c.json(await readAdminMetrics())
const snapshot = await readAdminMetrics()
deps.userMetricsRecorder.record(snapshot.value, snapshot.refreshedAt)
return c.json(snapshot.value)
})
.get('/users', async (c) => {