diff --git a/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus.types.ts b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/ReplicationPipelineStatus.types.ts
similarity index 62%
rename from apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus.types.ts
rename to apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/ReplicationPipelineStatus.types.ts
index f3c3cf2b8e..02aa0c694a 100644
--- a/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus.types.ts
+++ b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/ReplicationPipelineStatus.types.ts
@@ -3,6 +3,16 @@ export type RetryPolicy =
| { policy: 'manual_retry' }
| { policy: 'timed_retry'; next_retry: string }
+export type SlotLagMetrics = {
+ restart_lsn_bytes: number
+ confirmed_flush_lsn_bytes: number
+ safe_wal_size_bytes: number
+ write_lag?: number
+ flush_lag?: number
+}
+
+export type SlotLagMetricKey = keyof SlotLagMetrics
+
export type TableState = {
table_id: number
table_name: string
@@ -12,4 +22,5 @@ export type TableState = {
| { name: 'copied_table' }
| { name: 'following_wal'; lag: number }
| { name: 'error'; reason: string; solution?: string; retry_policy: RetryPolicy }
+ table_sync_lag?: SlotLagMetrics
}
diff --git a/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus.utils.tsx b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/ReplicationPipelineStatus.utils.tsx
similarity index 57%
rename from apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus.utils.tsx
rename to apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/ReplicationPipelineStatus.utils.tsx
index a4d3504431..76552b4d1d 100644
--- a/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus.utils.tsx
+++ b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/ReplicationPipelineStatus.utils.tsx
@@ -1,46 +1,57 @@
-import { ReplicationPipelineStatusData } from 'data/replication/pipeline-status-query'
+import dayjs from 'dayjs'
import { Activity, Clock, HelpCircle, Loader2, XCircle } from 'lucide-react'
+
+import { ReplicationPipelineStatusData } from 'data/replication/pipeline-status-query'
+import { formatBytes } from 'lib/helpers'
import { PipelineStatusRequestStatus } from 'state/replication-pipeline-request-status'
import { Badge } from 'ui'
-import { getPipelineStateMessages } from './Pipeline.utils'
+import { getPipelineStateMessages } from '../Pipeline.utils'
import { RetryPolicy, TableState } from './ReplicationPipelineStatus.types'
+const numberFormatter = new Intl.NumberFormat()
+
export const getStatusConfig = (state: TableState['state']) => {
switch (state.name) {
case 'queued':
return {
badge: Queued,
- description: 'Waiting to start replication',
+ description: 'Table is waiting for ETL to pick it up for replication.',
+ tooltip: 'Table is waiting for ETL to pick it up for replication.',
color: 'text-warning',
}
case 'copying_table':
return {
badge: Copying,
- description: 'Initial data copy in progress',
+ description: "Table's existing rows are being copied before live streaming begins.",
+ tooltip: "Table's existing rows are being copied before live streaming begins.",
color: 'text-brand-600',
}
case 'copied_table':
return {
badge: Copied,
- description: 'Initial copy completed',
+ description: "Table copy is complete and it's preparing to follow WAL changes.",
+ tooltip: "Table copy is complete and it's preparing to follow WAL changes.",
color: 'text-success-600',
}
case 'following_wal':
return {
badge: Live,
- description: `Replicating live changes`,
+ description: 'Table is streaming new changes in real time from the WAL.',
+ tooltip: 'Table is streaming new changes in real time from the WAL.',
color: 'text-success-600',
}
case 'error':
return {
badge: Error,
- description:
{state.reason}
,
+ description: 'Replication is paused because the table encountered an error.',
+ tooltip: 'Replication is paused because the table encountered an error.',
color: 'text-destructive-600',
}
default:
return {
badge: Unknown,
- description: 'Unknown status',
+ description: 'Table status is unavailable.',
+ tooltip: 'Table status is unavailable.',
color: 'text-warning',
}
}
@@ -122,3 +133,57 @@ export const isValidRetryPolicy = (policy: any): policy is RetryPolicy => {
return false
}
}
+
+const formatLagBytesValue = (value?: number) => {
+ if (typeof value !== 'number' || Number.isNaN(value)) {
+ return { display: '—', detail: undefined }
+ }
+
+ const decimals = value < 1024 ? 0 : value < 1024 * 1024 ? 1 : 2
+ const display = formatBytes(value, decimals)
+ const detail = `${numberFormatter.format(value)} bytes`
+
+ return { display, detail }
+}
+
+const formatLagDurationValue = (value?: number) => {
+ if (typeof value !== 'number' || Number.isNaN(value)) {
+ return { display: '—', detail: undefined }
+ }
+
+ const sign = value < 0 ? '-' : ''
+ const absMilliseconds = Math.abs(value)
+ const duration = dayjs.duration(absMilliseconds, 'milliseconds')
+
+ if (absMilliseconds < 1000) {
+ return { display: `${value} ms`, detail: undefined }
+ }
+
+ const seconds = duration.asSeconds()
+ if (seconds < 60) {
+ const decimals = seconds >= 10 ? 1 : 2
+ return {
+ display: `${sign}${seconds.toFixed(decimals)} s`,
+ detail: `${numberFormatter.format(value)} ms`,
+ }
+ }
+
+ const minutes = duration.asMinutes()
+ if (minutes < 60) {
+ const roundedSeconds = Math.round(seconds)
+ return {
+ display: `${sign}${minutes.toFixed(minutes >= 10 ? 1 : 2)} min`,
+ detail: `${numberFormatter.format(roundedSeconds)} s`,
+ }
+ }
+
+ const hours = duration.asHours()
+ const roundedMinutes = Math.round(minutes)
+ return {
+ display: `${sign}${hours.toFixed(hours >= 10 ? 1 : 2)} h`,
+ detail: `${numberFormatter.format(roundedMinutes)} min`,
+ }
+}
+
+export const getFormattedLagValue = (type: 'bytes' | 'duration', value?: number) =>
+ type === 'bytes' ? formatLagBytesValue(value) : formatLagDurationValue(value)
diff --git a/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/SlotLagMetrics.tsx b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/SlotLagMetrics.tsx
new file mode 100644
index 0000000000..1e11718fbe
--- /dev/null
+++ b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/SlotLagMetrics.tsx
@@ -0,0 +1,126 @@
+import { Info } from 'lucide-react'
+
+import { Tooltip, TooltipContent, TooltipTrigger } from 'ui'
+import { SlotLagMetricKey, SlotLagMetrics } from './ReplicationPipelineStatus.types'
+import { getFormattedLagValue } from './ReplicationPipelineStatus.utils'
+
+const SLOT_LAG_FIELDS: {
+ key: SlotLagMetricKey
+ label: string
+ type: 'bytes' | 'duration'
+ description: string
+}[] = [
+ {
+ key: 'confirmed_flush_lsn_bytes',
+ label: 'WAL Flush lag (size)',
+ type: 'bytes',
+ description:
+ 'Bytes between the newest WAL record applied locally and the latest flushed WAL record acknowledged by ETL.',
+ },
+ {
+ key: 'flush_lag',
+ label: 'WAL Flush lag (time)',
+ type: 'duration',
+ description:
+ 'Time between flushing recent WAL locally and receiving notification that ETL has written and flushed it.',
+ },
+ {
+ key: 'safe_wal_size_bytes',
+ label: 'Remaining WAL size',
+ type: 'bytes',
+ description:
+ 'Bytes still available to write to WAL before this slot risks entering the "lost" state.',
+ },
+]
+
+export const SlotLagMetricsInline = ({
+ tableName,
+ metrics,
+}: {
+ tableName: string
+ metrics: SlotLagMetrics
+}) => {
+ return (
+