import dayjs from 'dayjs' import { WARN_DURATION_ACTIVE_QUERY, WARN_DURATION_BLOCKED, WARN_DURATION_IDLE_TXN, WARN_TOP_BLOCKER, } from './DatabaseConnections.constants' import { type DatabaseActivity } from '@/data/database/activity-query' const LONG_RUNNING_STATES: (DatabaseActivity['state'] | undefined)[] = [ 'active', 'idle in transaction', 'idle in transaction (aborted)', ] type ActivityWithDuration = { activity: DatabaseActivity; duration: number } export type ConnectionMetrics = { activeQueries: DatabaseActivity[] blockedQueries: DatabaseActivity[] warnBlockedQueries: boolean longestBlockedQuery: ActivityWithDuration | null idleInTransactionQueries: DatabaseActivity[] longestRunningQuery: ActivityWithDuration | null warnLongestRunningQuery: boolean queryBlockingTheMostQueries: { activity: DatabaseActivity; count: number } | null warnTopBlocker: boolean } export const getDuration = (activity: DatabaseActivity) => { const { state } = activity if (state === 'active' && activity.query_start) { return dayjs().utc().diff(dayjs(activity.query_start).utc(), 'second') } if (state === 'idle' && activity.state_change) { return dayjs().utc().diff(dayjs(activity.state_change).utc(), 'second') } if ( (state === 'idle in transaction' || state === 'idle in transaction (aborted)') && activity.transaction_start ) { return dayjs().utc().diff(dayjs(activity.transaction_start).utc(), 'second') } return null } export const getBadgeVariant = (activity: DatabaseActivity) => { const { state } = activity if (state === 'active') return 'success' if (state === 'idle in transaction' || state === 'idle in transaction (aborted)') return 'warning' return 'default' } // Resolves the timestamp that duration should be measured from for a given state - // query_start while actively running, transaction_start while idle in transaction. export const getActivityStart = (activity: DatabaseActivity) => { if (activity.state === 'active') return activity.query_start if ( activity.state === 'idle in transaction' || activity.state === 'idle in transaction (aborted)' ) { return activity.transaction_start } return null } // Finds the activity with the largest duration-since-start among the given activities (e.g. the // longest-blocked query, or the longest-running query). const findLongestRunning = ( activities: DatabaseActivity[], now: dayjs.Dayjs ): ActivityWithDuration | null => activities.reduce((longest, activity) => { const start = getActivityStart(activity) if (!start) return longest const duration = Math.max(now.diff(dayjs(start).utc(), 'second'), 0) return longest === null || duration > longest.duration ? { activity, duration } : longest }, null) // Counts every activity transitively blocked by pid - not just direct waiters, but their waiters // in turn - so a long chain outweighs several short ones. Traverses breadth-first over a single // counted set so a waiter reachable through more than one blocker (a "diamond") is still only // counted once. const countTransitivelyBlocked = (pid: number, activities: DatabaseActivity[]): number => { const counted = new Set() const queue = [pid] while (queue.length > 0) { const currentPid = queue.shift()! const directWaiters = activities.filter((x) => x.blocked_by.includes(currentPid)) for (const waiter of directWaiters) { if (counted.has(waiter.pid)) continue counted.add(waiter.pid) queue.push(waiter.pid) } } return counted.size } // Derives the Overview page's connection/activity metrics (and their warning thresholds) from the // raw pg_stat_activity rows, so the logic can be unit tested independently of the component. export const getConnectionMetrics = (activities: DatabaseActivity[]): ConnectionMetrics => { // Computed once so every metric below is measured against the same instant, rather than each // drifting slightly against the others. const now = dayjs().utc() const activeQueries = activities.filter((x) => x.state === 'active') const blockedQueries = activities.filter((x) => x.blocked_by.length > 0) const warnBlockedQueries = blockedQueries.some((activity) => { const start = getActivityStart(activity) if (!start) return false return now.diff(dayjs(start).utc(), 'second') >= WARN_DURATION_BLOCKED }) const longestBlockedQuery = findLongestRunning(blockedQueries, now) const idleInTransactionQueries = activities.filter((x) => { const isIdleInTransaction = x.state === 'idle in transaction' || x.state === 'idle in transaction (aborted)' if (!isIdleInTransaction || !x.transaction_start) return false return now.diff(dayjs(x.transaction_start).utc(), 'second') >= WARN_DURATION_IDLE_TXN }) const longestRunningQuery = findLongestRunning( activities.filter((x) => LONG_RUNNING_STATES.includes(x.state)), now ) const warnLongestRunningQuery = (longestRunningQuery?.activity.state === 'active' && longestRunningQuery.duration >= WARN_DURATION_ACTIVE_QUERY) || ((longestRunningQuery?.activity.state === 'idle in transaction' || longestRunningQuery?.activity.state === 'idle in transaction (aborted)') && longestRunningQuery.duration >= WARN_DURATION_IDLE_TXN) const queryBlockingTheMostQueries = activities.reduce<{ activity: DatabaseActivity count: number } | null>((mostBlocking, activity) => { const count = countTransitivelyBlocked(activity.pid, activities) if (count === 0) return mostBlocking if (mostBlocking && count <= mostBlocking.count) return mostBlocking return { activity, count } }, null) const warnTopBlocker = (queryBlockingTheMostQueries?.count ?? 0) >= WARN_TOP_BLOCKER return { activeQueries, blockedQueries, warnBlockedQueries, longestBlockedQuery, idleInTransactionQueries, longestRunningQuery, warnLongestRunningQuery, queryBlockingTheMostQueries, warnTopBlocker, } } export type ActivityFilters = { search: string states: string[] applications: string[] roles: string[] view: string } // Applies the Sessions table's filters to the raw pg_stat_activity rows. Shared so that anything // selecting a pid (e.g. the Overview metric cards) can check whether that pid would actually be // visible under the current filters, without duplicating this predicate. export const filterActivities = ( activities: DatabaseActivity[], filters: ActivityFilters ): DatabaseActivity[] => { const { search, states, applications, roles, view } = filters // Pids referenced in some other activity's blocked_by - i.e. they are blocking something const blockingPids = new Set(activities.flatMap((x) => x.blocked_by)) return activities.filter((activity) => { const matchesState = states.length === 0 || (activity.state !== null && states.includes(activity.state)) const matchesRole = roles.length === 0 || roles.includes(activity.role_name) const matchesApplication = applications.length === 0 || applications.includes(activity.application_name) // In the blocked view, only show root blockers - activities blocking others while not // themselves blocked. Everything they block is shown nested under them instead. const matchesView = view !== 'blockers' || (activity.blocked_by.length === 0 && blockingPids.has(activity.pid)) const matchesSearch = !search || (activity.query?.toLowerCase().includes(search.toLowerCase()) ?? false) return matchesState && matchesRole && matchesApplication && matchesView && matchesSearch }) } export const getBlockChain = (pid: number, activities: DatabaseActivity[]) => { const chain = [pid] const visited = new Set([pid]) let current = activities.find((x) => x.pid === pid) while (current && current.blocked_by.length > 0) { const nextPid = current.blocked_by[0] if (visited.has(nextPid)) break chain.push(nextPid) visited.add(nextPid) current = activities.find((x) => x.pid === nextPid) } return chain } // Walks the opposite direction of getBlockChain: starting from a root blocker // (blocked_by.length === 0), finds the chain of pids waiting on it, nearest first export const getBlockingChain = (rootPid: number, activities: DatabaseActivity[]) => { const chain: number[] = [] const visited = new Set([rootPid]) let currentPid = rootPid while (true) { const next = activities.find((x) => !visited.has(x.pid) && x.blocked_by.includes(currentPid)) if (!next) break chain.push(next.pid) visited.add(next.pid) currentPid = next.pid } return chain }