@@ -298,8 +484,8 @@ displayStore.defaultBack = '/dashboard'
{{ slot.status.toUpperCase() }}
diff --git a/supabase/functions/_backend/public/replication.ts b/supabase/functions/_backend/public/replication.ts
index 8aa9bc8085..21f861c774 100644
--- a/supabase/functions/_backend/public/replication.ts
+++ b/supabase/functions/_backend/public/replication.ts
@@ -1,4 +1,6 @@
+import type { Context } from 'hono'
import { sql } from 'drizzle-orm'
+import { CacheHelper } from '../utils/cache.ts'
import { honoFactory, useCors } from '../utils/hono.ts'
import { cloudlogErr } from '../utils/logging.ts'
import { closeClient, getDrizzleClient, getPgClient, logPgError } from '../utils/pg.ts'
@@ -6,8 +8,14 @@ import { validatePlatformAdminOrApiSecret } from '../utils/platform_admin_access
const DEFAULT_THRESHOLD_SECONDS = 180
const DEFAULT_THRESHOLD_BYTES = 16 * 1024 * 1024
+const DATA_CANARY_TABLE = 'app_versions'
+const DATA_CANARY_TTL_SECONDS = 300
+const DATA_CANARY_TTL_MS = DATA_CANARY_TTL_SECONDS * 1000
+const DATA_CANARY_THRESHOLD_PERCENT = 0.01
+const DATA_CANARY_CACHE_TIMEOUT_MS = 250
type SlotStatus = 'ok' | 'ko'
+type CheckStatus = 'ok' | 'ko' | 'skipped'
type ReplicationQueryMode = 'wal_stats' | 'replication_stats' | 'slots_only'
interface ReplicationSlotLag {
@@ -25,6 +33,53 @@ interface ReplicationSlotLag {
reasons: string[]
}
+export interface SubscriptionWorkerRow {
+ subname: string
+ subenabled: boolean
+ has_apply_worker: boolean
+ has_recent_receipt: boolean
+ apply_lag_seconds: number | null
+ last_msg_receipt_time: string | null
+}
+
+export interface SubscriptionHealthResult {
+ status: CheckStatus
+ checked_at: string
+ threshold_seconds: number
+ subscriptions: Array<{
+ subname: string
+ enabled: boolean
+ has_apply_worker: boolean
+ has_recent_receipt: boolean
+ apply_lag_seconds: number | null
+ last_msg_receipt_time: string | null
+ status: SlotStatus | 'disabled'
+ reasons: string[]
+ }>
+ reasons: string[]
+}
+
+export interface DataCanaryResult {
+ status: CheckStatus
+ table: typeof DATA_CANARY_TABLE
+ primary_count: number | null
+ replica_count: number | null
+ diff: number | null
+ diff_percent: number | null
+ threshold_percent: number
+ checked_at: string
+ expires_at: string
+ cached: boolean
+ reasons: string[]
+}
+
+interface DataCanaryCacheEntry extends Omit {
+ expiresAt: number
+}
+
+const dataCanaryMemoryCache = new Map()
+const dataCanaryInflight = new Map>()
+
function toNumber(value: unknown): number | null {
if (value === null || value === undefined)
return null
@@ -34,6 +89,120 @@ function toNumber(value: unknown): number | null {
return num
}
+function isReplicaDatabaseSource(source: string): boolean {
+ return source.startsWith('HYPERDRIVE_CAPGO_READ') || source === 'local_read_replica'
+}
+
+export function evaluateAppVersionsCanary(
+ primaryCount: number,
+ replicaCount: number,
+ thresholdPercent = DATA_CANARY_THRESHOLD_PERCENT,
+): Pick {
+ const diff = Math.abs(primaryCount - replicaCount)
+ const baseline = Math.max(primaryCount, replicaCount, 0)
+ const diffPercent = baseline === 0 ? 0 : diff / baseline
+ const reasons: string[] = []
+
+ if (replicaCount === 0)
+ reasons.push('replica_empty')
+ else if (diffPercent > thresholdPercent)
+ reasons.push('count_mismatch')
+
+ return {
+ status: reasons.length > 0 ? 'ko' : 'ok',
+ diff,
+ diff_percent: Number(diffPercent.toFixed(6)),
+ reasons,
+ }
+}
+
+export function evaluateSubscriptionHealth(
+ rows: SubscriptionWorkerRow[],
+ thresholdSeconds = DEFAULT_THRESHOLD_SECONDS,
+ checkedAt = new Date().toISOString(),
+): SubscriptionHealthResult {
+ if (rows.length === 0) {
+ return {
+ status: 'ko',
+ checked_at: checkedAt,
+ threshold_seconds: thresholdSeconds,
+ subscriptions: [],
+ reasons: ['no_subscription'],
+ }
+ }
+
+ const subscriptions = rows.map((row) => {
+ if (!row.subenabled) {
+ return {
+ subname: row.subname,
+ enabled: false,
+ has_apply_worker: row.has_apply_worker,
+ has_recent_receipt: row.has_recent_receipt,
+ apply_lag_seconds: row.apply_lag_seconds,
+ last_msg_receipt_time: row.last_msg_receipt_time,
+ status: 'disabled' as const,
+ reasons: ['subscription_disabled'],
+ }
+ }
+
+ const reasons: string[] = []
+ if (!row.has_apply_worker)
+ reasons.push('no_apply_worker')
+ if (!row.has_recent_receipt)
+ reasons.push('no_recent_receipt')
+ if (row.apply_lag_seconds !== null && row.apply_lag_seconds > thresholdSeconds)
+ reasons.push('apply_lag_threshold_exceeded')
+
+ return {
+ subname: row.subname,
+ enabled: true,
+ has_apply_worker: row.has_apply_worker,
+ has_recent_receipt: row.has_recent_receipt,
+ apply_lag_seconds: row.apply_lag_seconds,
+ last_msg_receipt_time: row.last_msg_receipt_time,
+ status: (reasons.length > 0 ? 'ko' : 'ok') as SlotStatus,
+ reasons,
+ }
+ })
+
+ // Disabled leftovers are ignored. Every enabled subscription must be healthy.
+ const enabled = subscriptions.filter(sub => sub.enabled)
+ const allEnabledHealthy = enabled.length > 0 && enabled.every(sub => sub.status === 'ok')
+ const reasons = allEnabledHealthy
+ ? []
+ : [...new Set((enabled.length ? enabled : subscriptions).flatMap(sub => sub.reasons))]
+
+ return {
+ status: allEnabledHealthy ? 'ok' : 'ko',
+ checked_at: checkedAt,
+ threshold_seconds: thresholdSeconds,
+ subscriptions,
+ reasons: reasons.length > 0 ? reasons : ['subscription_unhealthy'],
+ }
+}
+
+function getFreshDataCanaryMemoryEntry(cacheKey: string, now = Date.now()): DataCanaryResult | null {
+ const cached = dataCanaryMemoryCache.get(cacheKey)
+ if (!cached)
+ return null
+ if (cached.expiresAt <= now) {
+ dataCanaryMemoryCache.delete(cacheKey)
+ return null
+ }
+ const { expiresAt: _expiresAt, ...payload } = cached
+ return { ...payload, cached: true }
+}
+
+function setDataCanaryMemoryEntry(cacheKey: string, entry: DataCanaryCacheEntry) {
+ dataCanaryMemoryCache.set(cacheKey, entry)
+}
+
+/** Test helper: clear in-process canary cache between unit tests. */
+export function clearDataCanaryCacheForTests() {
+ dataCanaryMemoryCache.clear()
+ dataCanaryInflight.clear()
+}
+
function buildReplicationQuery(mode: ReplicationQueryMode) {
const slotsCte = sql`
WITH slots AS (
@@ -139,6 +308,163 @@ async function executeReplicationQuery(
throw lastError
}
+async function querySubscriptionHealth(
+ drizzleClient: ReturnType,
+ thresholdSeconds: number,
+): Promise {
+ const checkedAt = new Date().toISOString()
+ const result = await drizzleClient.execute(sql`
+ SELECT
+ s.subname,
+ s.subenabled,
+ COALESCE(bool_or(ss.pid IS NOT NULL), false) AS has_apply_worker,
+ COALESCE(bool_or(ss.last_msg_receipt_time IS NOT NULL), false) AS has_recent_receipt,
+ MAX(EXTRACT(EPOCH FROM (now() - ss.last_msg_receipt_time)))
+ FILTER (WHERE ss.last_msg_receipt_time IS NOT NULL) AS apply_lag_seconds,
+ MAX(ss.last_msg_receipt_time) AS last_msg_receipt_time
+ FROM pg_subscription s
+ LEFT JOIN pg_stat_subscription ss ON ss.subname = s.subname
+ GROUP BY s.subname, s.subenabled
+ ORDER BY s.subname
+ `)
+
+ const rows: SubscriptionWorkerRow[] = (result.rows as any[]).map(row => ({
+ subname: String(row.subname),
+ subenabled: Boolean(row.subenabled),
+ has_apply_worker: Boolean(row.has_apply_worker),
+ has_recent_receipt: Boolean(row.has_recent_receipt),
+ apply_lag_seconds: toNumber(row.apply_lag_seconds),
+ last_msg_receipt_time: row.last_msg_receipt_time
+ ? new Date(row.last_msg_receipt_time).toISOString()
+ : null,
+ }))
+
+ return evaluateSubscriptionHealth(rows, thresholdSeconds, checkedAt)
+}
+
+async function countAppVersions(drizzleClient: ReturnType): Promise {
+ const result = await drizzleClient.execute(sql`
+ SELECT COUNT(*)::bigint AS count
+ FROM public.app_versions
+ `)
+ const count = toNumber((result.rows as any[])[0]?.count)
+ if (count === null)
+ throw new Error('app_versions count missing')
+ return count
+}
+
+async function queryDataCanary(
+ primaryDrizzle: ReturnType,
+ replicaDrizzle: ReturnType,
+): Promise & { expiresAt: number }> {
+ const checkedAt = new Date().toISOString()
+ const expiresAt = Date.now() + DATA_CANARY_TTL_MS
+ const [primaryCount, replicaCount] = await Promise.all([
+ countAppVersions(primaryDrizzle),
+ countAppVersions(replicaDrizzle),
+ ])
+ const evaluation = evaluateAppVersionsCanary(primaryCount, replicaCount)
+
+ return {
+ status: evaluation.status,
+ table: DATA_CANARY_TABLE,
+ primary_count: primaryCount,
+ replica_count: replicaCount,
+ diff: evaluation.diff,
+ diff_percent: evaluation.diff_percent,
+ threshold_percent: DATA_CANARY_THRESHOLD_PERCENT,
+ checked_at: checkedAt,
+ expiresAt,
+ reasons: evaluation.reasons,
+ }
+}
+
+async function getCachedDataCanary(
+ c: Context,
+ primaryDrizzle: ReturnType,
+ replicaDrizzle: ReturnType,
+ replicaSource: string,
+): Promise {
+ const cacheKey = `app_versions:${replicaSource}`
+ const memoryEntry = getFreshDataCanaryMemoryEntry(cacheKey)
+ if (memoryEntry)
+ return memoryEntry
+
+ const existingQuery = dataCanaryInflight.get(cacheKey)
+ if (existingQuery)
+ return existingQuery
+
+ const cacheHelper = new CacheHelper(c)
+ const cacheRequest = cacheHelper.buildRequest('/cache/replication-data-canary', { source: cacheKey })
+ const cachedEntry = await cacheHelper.matchJson(cacheRequest, {
+ timeoutMs: DATA_CANARY_CACHE_TIMEOUT_MS,
+ })
+
+ if (cachedEntry && cachedEntry.expiresAt > Date.now()) {
+ setDataCanaryMemoryEntry(cacheKey, cachedEntry)
+ const { expiresAt: _expiresAt, ...payload } = cachedEntry
+ return { ...payload, cached: true }
+ }
+
+ const existingQueryAfterCache = dataCanaryInflight.get(cacheKey)
+ if (existingQueryAfterCache)
+ return existingQueryAfterCache
+
+ const query = queryDataCanary(primaryDrizzle, replicaDrizzle)
+ .then(async (result) => {
+ const cacheEntry: DataCanaryCacheEntry = {
+ status: result.status,
+ table: result.table,
+ primary_count: result.primary_count,
+ replica_count: result.replica_count,
+ diff: result.diff,
+ diff_percent: result.diff_percent,
+ threshold_percent: result.threshold_percent,
+ checked_at: result.checked_at,
+ expires_at: new Date(result.expiresAt).toISOString(),
+ expiresAt: result.expiresAt,
+ reasons: result.reasons,
+ }
+ setDataCanaryMemoryEntry(cacheKey, cacheEntry)
+ await cacheHelper.putJson(cacheRequest, cacheEntry, DATA_CANARY_TTL_SECONDS, { timeoutMs: DATA_CANARY_CACHE_TIMEOUT_MS })
+ const { expiresAt: _expiresAt, ...payload } = cacheEntry
+ return { ...payload, cached: false }
+ })
+ .finally(() => {
+ dataCanaryInflight.delete(cacheKey)
+ })
+
+ dataCanaryInflight.set(cacheKey, query)
+ return query
+}
+
+function skippedSubscription(reason: string): SubscriptionHealthResult {
+ return {
+ status: 'skipped',
+ checked_at: new Date().toISOString(),
+ threshold_seconds: DEFAULT_THRESHOLD_SECONDS,
+ subscriptions: [],
+ reasons: [reason],
+ }
+}
+
+function skippedDataCanary(reason: string): DataCanaryResult {
+ const checkedAt = new Date().toISOString()
+ return {
+ status: 'skipped',
+ table: DATA_CANARY_TABLE,
+ primary_count: null,
+ replica_count: null,
+ diff: null,
+ diff_percent: null,
+ threshold_percent: DATA_CANARY_THRESHOLD_PERCENT,
+ checked_at: checkedAt,
+ expires_at: checkedAt,
+ cached: false,
+ reasons: [reason],
+ }
+}
+
export const app = honoFactory.createApp()
app.use('*', useCors)
@@ -154,6 +480,7 @@ app.get('/', async (c) => {
const pgClient = getPgClient(c, false)
const drizzleClient = getDrizzleClient(pgClient)
+ let replicaPgClient: ReturnType | null = null
try {
const { rows, mode } = await executeReplicationQuery({ requestId: c.get('requestId') }, drizzleClient)
@@ -212,10 +539,50 @@ app.get('/', async (c) => {
return acc
}, { slot: null, lag: null })
- const overallStatus: SlotStatus = slots.length === 0 || slots.some(slot => slot.status === 'ko') ? 'ko' : 'ok'
+ let subscription = skippedSubscription('no_replica_connection')
+ let dataCanary = skippedDataCanary('no_replica_connection')
+
+ try {
+ replicaPgClient = getPgClient(c, true)
+ const replicaSource = c.res.headers.get('X-Database-Source') ?? ''
+ if (!isReplicaDatabaseSource(replicaSource)) {
+ subscription = skippedSubscription('no_replica_connection')
+ dataCanary = skippedDataCanary('no_replica_connection')
+ }
+ else {
+ const replicaDrizzle = getDrizzleClient(replicaPgClient)
+ const [subscriptionResult, canaryResult] = await Promise.all([
+ querySubscriptionHealth(replicaDrizzle, thresholdSeconds),
+ getCachedDataCanary(c, drizzleClient, replicaDrizzle, replicaSource),
+ ])
+ subscription = subscriptionResult
+ dataCanary = canaryResult
+ }
+ }
+ catch (error) {
+ cloudlogErr({ requestId: c.get('requestId'), message: 'replication_replica_check_failed', error })
+ subscription = {
+ ...skippedSubscription('replica_check_failed'),
+ status: 'ko',
+ }
+ dataCanary = {
+ ...skippedDataCanary('replica_check_failed'),
+ status: 'ko',
+ }
+ }
+
+ const slotStatus: SlotStatus = slots.length === 0 || slots.some(slot => slot.status === 'ko') ? 'ko' : 'ok'
+ const failingChecks = [
+ slotStatus === 'ko',
+ subscription.status === 'ko',
+ dataCanary.status === 'ko',
+ ]
+ // Intentionally includes subscription + canary: /replication is the admin health probe.
+ const overallStatus: SlotStatus = failingChecks.some(Boolean) ? 'ko' : 'ok'
const response = {
status: overallStatus,
+ slot_status: slotStatus,
estimation_source: mode,
threshold_seconds: thresholdSeconds,
threshold_minutes: Number((thresholdSeconds / 60).toFixed(2)),
@@ -228,6 +595,8 @@ app.get('/', async (c) => {
max_lag_minutes: maxLagSlot.lag !== null ? Number((maxLagSlot.lag / 60).toFixed(2)) : null,
max_lag_slot: maxLagSlot.slot,
slots,
+ subscription,
+ data_canary: dataCanary,
}
return c.json(response, overallStatus === 'ok' ? 200 : 503)
@@ -250,9 +619,14 @@ app.get('/', async (c) => {
max_lag_minutes: null,
max_lag_slot: null,
slots: [],
+ slot_status: 'ko',
+ subscription: skippedSubscription('replication_lag_error'),
+ data_canary: skippedDataCanary('replication_lag_error'),
}, 500)
}
finally {
await closeClient(c, pgClient)
+ if (replicaPgClient)
+ await closeClient(c, replicaPgClient)
}
})
diff --git a/tests/replication-data-canary.unit.test.ts b/tests/replication-data-canary.unit.test.ts
new file mode 100644
index 0000000000..487b6705fa
--- /dev/null
+++ b/tests/replication-data-canary.unit.test.ts
@@ -0,0 +1,138 @@
+import { afterEach, describe, expect, it } from 'vitest'
+import {
+ clearDataCanaryCacheForTests,
+ evaluateAppVersionsCanary,
+ evaluateSubscriptionHealth,
+} from '../supabase/functions/_backend/public/replication.ts'
+
+describe('replication data canary evaluation', () => {
+ afterEach(() => {
+ clearDataCanaryCacheForTests()
+ })
+
+ it('accepts similar app_versions counts within 1%', () => {
+ expect(evaluateAppVersionsCanary(1000, 995)).toMatchObject({
+ status: 'ok',
+ diff: 5,
+ reasons: [],
+ })
+ })
+
+ it('rejects empty replica when primary has rows', () => {
+ expect(evaluateAppVersionsCanary(120, 0)).toMatchObject({
+ status: 'ko',
+ reasons: ['replica_empty'],
+ })
+ })
+
+ it('rejects empty replica even when primary is empty', () => {
+ expect(evaluateAppVersionsCanary(0, 0)).toMatchObject({
+ status: 'ko',
+ reasons: ['replica_empty'],
+ })
+ })
+
+ it('rejects count drift above threshold', () => {
+ expect(evaluateAppVersionsCanary(1000, 900)).toMatchObject({
+ status: 'ko',
+ reasons: ['count_mismatch'],
+ })
+ })
+
+ it('treats one healthy subscription as ok even with a disabled sibling', () => {
+ const result = evaluateSubscriptionHealth([
+ {
+ subname: 'capgo_google_eu_2',
+ subenabled: false,
+ has_apply_worker: false,
+ has_recent_receipt: false,
+ apply_lag_seconds: null,
+ last_msg_receipt_time: null,
+ },
+ {
+ subname: 'capgo_google_eu_2_sub',
+ subenabled: true,
+ has_apply_worker: true,
+ has_recent_receipt: true,
+ apply_lag_seconds: 2,
+ last_msg_receipt_time: '2026-08-01T07:00:00.000Z',
+ },
+ ])
+
+ expect(result.status).toBe('ok')
+ expect(result.subscriptions.find(s => s.subname === 'capgo_google_eu_2')?.status).toBe('disabled')
+ expect(result.subscriptions.find(s => s.subname === 'capgo_google_eu_2_sub')?.status).toBe('ok')
+ })
+
+ it('fails when any enabled subscription is unhealthy', () => {
+ const result = evaluateSubscriptionHealth([
+ {
+ subname: 'capgo_google_eu_2',
+ subenabled: true,
+ has_apply_worker: false,
+ has_recent_receipt: false,
+ apply_lag_seconds: null,
+ last_msg_receipt_time: null,
+ },
+ {
+ subname: 'capgo_google_eu_2_sub',
+ subenabled: true,
+ has_apply_worker: true,
+ has_recent_receipt: true,
+ apply_lag_seconds: 2,
+ last_msg_receipt_time: '2026-08-01T07:00:00.000Z',
+ },
+ ])
+
+ expect(result.status).toBe('ko')
+ expect(result.reasons).toContain('no_apply_worker')
+ })
+
+ it('marks enabled subscription without apply worker as ko', () => {
+ const result = evaluateSubscriptionHealth([
+ {
+ subname: 'capgo_google_eu_2_sub',
+ subenabled: true,
+ has_apply_worker: false,
+ has_recent_receipt: false,
+ apply_lag_seconds: null,
+ last_msg_receipt_time: null,
+ },
+ ])
+
+ expect(result.status).toBe('ko')
+ expect(result.reasons).toContain('no_apply_worker')
+ })
+
+ it('marks enabled subscription with pid but no receipt as ko', () => {
+ const result = evaluateSubscriptionHealth([
+ {
+ subname: 'capgo_google_eu_2_sub',
+ subenabled: true,
+ has_apply_worker: true,
+ has_recent_receipt: false,
+ apply_lag_seconds: null,
+ last_msg_receipt_time: null,
+ },
+ ])
+
+ expect(result.status).toBe('ko')
+ expect(result.reasons).toContain('no_recent_receipt')
+ })
+
+ it('marks apply lag above threshold as ko', () => {
+ const result = evaluateSubscriptionHealth([
+ {
+ subname: 'capgo_google_eu_2_sub',
+ subenabled: true,
+ has_apply_worker: true,
+ has_recent_receipt: true,
+ apply_lag_seconds: 400,
+ last_msg_receipt_time: '2026-08-01T06:50:00.000Z',
+ },
+ ], 180)
+
+ expect(result.status).toBe('ko')
+ expect(result.reasons).toContain('apply_lag_threshold_exceeded')
+ })
+})
|