From 7ab3f326253e42c0c0e7c9b2d977610d9ff86a4a Mon Sep 17 00:00:00 2001 From: Danny White <3104761+dnywh@users.noreply.github.com> Date: Thu, 17 Sep 2026 16:31:04 +1000 Subject: [PATCH] feat(studio): rebuild the pipeline overview (#49630) ## What kind of change does this PR introduce? Studio UI improvement. ## What is the current behavior? The pipeline Overview uses bespoke loading, metrics, table-state and empty-state layouts that shift while data resolves and repeat status information from the detail header. ## What is the new behavior? Rebuilds the Overview around stable **Pipeline health** and **Replicated tables** sections. It adds layout-matched loading geometry, prioritised pipeline notices, initial-sync progress, clearer empty states, and accessible loading announcements. Complete pipeline configuration remains deferred to #49631. | Before | After | | --- | --- | | 54861 | Replication Database Agua Basket Supabase | This is the final independent slice in the review series: #50443, #50444, #50445, #50446, then this PR. Each PR targets `master` and can merge on its own. Rebase this PR as earlier slices merge. ## To test 1. Open `/project//database/replication` and select a pipeline. 2. Throttle the initial requests and confirm **Pipeline health** and **Replicated tables** keep their final geometry while loading. 3. Check running, initial-sync, stopped, failed, disconnected and unavailable states. 4. Confirm the Overview contains Pipeline health and Replicated tables only, without a Configuration section. ## Summary by CodeRabbit - **New Features** - Reorganized replication pipeline status into Pipeline health and Replicated tables sections. - Added loading skeletons with accessible status announcements. - Added clearer notices for pipeline health, failed or disconnected pipelines, paused updates, lag, and synchronization progress. - Improved empty states when table data is unavailable or the pipeline is inactive. - Added options to view logs and reset failed tables. - **Tests** - Added coverage for loading behavior, pipeline notices, table counts, synchronization progress, and empty states. --- .../PipelineOverview.utils.test.ts | 94 +++ .../PipelineOverview.utils.ts | 126 ++++ .../ReplicationPipelineStatus.test.tsx | 95 +++ .../ReplicationPipelineStatus.tsx | 551 ++++++++++-------- .../TableReplicationRow.tsx | 2 +- .../Database/Replication/RowMenu.tsx | 2 +- 6 files changed, 630 insertions(+), 240 deletions(-) create mode 100644 apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/PipelineOverview.utils.test.ts create mode 100644 apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/PipelineOverview.utils.ts create mode 100644 apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/ReplicationPipelineStatus.test.tsx diff --git a/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/PipelineOverview.utils.test.ts b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/PipelineOverview.utils.test.ts new file mode 100644 index 00000000000..b14e4fdccdf --- /dev/null +++ b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/PipelineOverview.utils.test.ts @@ -0,0 +1,94 @@ +import { describe, expect, test } from 'vitest' + +import { + getInitialSyncSummary, + getPipelineStateNotice, + getTableStatusEmptyState, +} from './PipelineOverview.utils' +import { PipelineStatusRequestStatus } from '@/state/replication-pipeline-request-status' + +const disabledStateConfig = { title: 'Starting pipeline', message: 'This can take a moment.' } +const liveTables = (count: number) => + Array.from({ length: count }, () => ({ state: { name: 'following_wal' as const } })) + +describe('getTableStatusEmptyState', () => { + test.each([ + [true, 'stopped' as const, 'Starting pipeline'], + [false, 'stopped' as const, 'Pipeline stopped'], + [false, 'failed' as const, 'Pipeline failed'], + [false, 'started' as const, 'No table data yet'], + ])('returns the appropriate empty state', (isDisabled, statusName, title) => { + expect(getTableStatusEmptyState({ isDisabled, disabledStateConfig, statusName }).title).toBe( + title + ) + }) +}) + +describe('getPipelineStateNotice', () => { + test('omits a notice for a healthy running pipeline', () => { + expect( + getPipelineStateNotice({ + requestStatus: PipelineStatusRequestStatus.None, + statusName: 'started', + tableStatuses: liveTables(3), + }) + ).toBeUndefined() + }) + + test.each([ + ['failed' as const, 'destructive', true], + ['stopped' as const, 'note', false], + ])('explains a %s pipeline', (statusName, type, showLogsLink) => { + expect( + getPipelineStateNotice({ + requestStatus: PipelineStatusRequestStatus.None, + statusName, + tableStatuses: liveTables(3), + }) + ).toMatchObject({ type, showLogsLink }) + }) + + test('reports a requested transition ahead of the API status', () => { + expect( + getPipelineStateNotice({ + requestStatus: PipelineStatusRequestStatus.StartRequested, + statusName: 'stopped', + tableStatuses: liveTables(3), + })?.title + ).toBe('Starting pipeline') + }) + + test('distinguishes copying tables from queued tables', () => { + expect( + getPipelineStateNotice({ + requestStatus: PipelineStatusRequestStatus.None, + statusName: 'started', + tableStatuses: [ + ...liveTables(2), + { state: { name: 'copying_table' as const } }, + { state: { name: 'queued' as const } }, + ], + })?.description + ).toContain('1 of 4 tables is copying and 1 is waiting.') + }) +}) + +describe('getInitialSyncSummary', () => { + test.each([ + [4, 3, 8, '4 of 8 tables are copying and 3 are waiting.'], + [1, 1, 4, '1 of 4 tables is copying and 1 is waiting.'], + [2, 0, 8, '2 of 8 tables are copying.'], + [0, 3, 3, '3 tables are waiting to copy.'], + [1, 0, 1, '1 of 1 table is copying.'], + [0, 0, 2, 'The last tables are finishing their copy.'], + ])('summarises initial sync progress', (copyingCount, queuedCount, totalCount, expected) => { + expect( + getInitialSyncSummary({ + syncingCount: copyingCount + queuedCount, + copyingCount, + queuedCount, + totalCount, + }) + ).toBe(expected) + }) +}) diff --git a/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/PipelineOverview.utils.ts b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/PipelineOverview.utils.ts new file mode 100644 index 00000000000..70a0431d63a --- /dev/null +++ b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/PipelineOverview.utils.ts @@ -0,0 +1,126 @@ +import { getPipelineDisplayState, normalizePipelineStatusName } from '../Pipeline.utils' +import { PipelineStatusName } from '../Replication.constants' +import { TableState } from './ReplicationPipelineStatus.types' +import { getInitialSyncProgress } from './ReplicationPipelineStatus.utils' +import { ReplicationPipelineStatusData } from '@/data/replication/pipeline-status-query' +import { PipelineStatusRequestStatus } from '@/state/replication-pipeline-request-status' + +export const getTableStatusEmptyState = ({ + isDisabled, + disabledStateConfig, + statusName, +}: { + isDisabled: boolean + disabledStateConfig: { title: string; message: string } + statusName?: ReplicationPipelineStatusData['status']['name'] +}) => { + if (isDisabled) { + return { title: disabledStateConfig.title, description: disabledStateConfig.message } + } + + if (statusName === PipelineStatusName.STOPPED) { + return { title: 'Pipeline stopped', description: 'Start the pipeline to begin replication' } + } + + if (statusName === PipelineStatusName.FAILED) { + return { + title: 'Pipeline failed', + description: 'Restart the pipeline or reset your tables to recover', + } + } + + return { + title: 'No table data yet', + description: 'Table status appears here once replication begins', + } +} + +export interface PipelineStateNotice { + type: 'note' | 'warning' | 'destructive' + title: string + description: string + showLogsLink: boolean +} + +const plural = (count: number, singular: string, pluralForm = `${singular}s`) => + `${count} ${count === 1 ? singular : pluralForm}` + +export const getInitialSyncSummary = ({ + copyingCount, + queuedCount, + totalCount, +}: ReturnType) => { + if (copyingCount > 0 && queuedCount > 0) { + return `${copyingCount} of ${plural(totalCount, 'table')} ${copyingCount === 1 ? 'is' : 'are'} copying and ${queuedCount} ${queuedCount === 1 ? 'is' : 'are'} waiting.` + } + if (copyingCount > 0) { + return `${copyingCount} of ${plural(totalCount, 'table')} ${copyingCount === 1 ? 'is' : 'are'} copying.` + } + if (queuedCount > 0) { + return `${plural(queuedCount, 'table')} ${queuedCount === 1 ? 'is' : 'are'} waiting to copy.` + } + return 'The last tables are finishing their copy.' +} + +export const getPipelineStateNotice = ({ + requestStatus, + statusName, + tableStatuses, +}: { + requestStatus: PipelineStatusRequestStatus + statusName?: ReplicationPipelineStatusData['status']['name'] + tableStatuses: { state: { name: TableState['state']['name'] } }[] +}): PipelineStateNotice | undefined => { + const displayState = getPipelineDisplayState( + requestStatus, + normalizePipelineStatusName(statusName) + ) + + if (displayState.type === 'loading') { + return { + type: 'note', + title: displayState.title, + description: displayState.message, + showLogsLink: false, + } + } + + if (displayState.key === 'failed') { + return { + type: 'destructive', + title: displayState.title, + description: + 'Replication has stopped. Restart the pipeline to resume from its last checkpoint. Table states below are from before it failed.', + showLogsLink: true, + } + } + + if (displayState.key === 'stopped') { + return { + type: 'note', + title: displayState.title, + description: + 'Changes to your source tables wait in Postgres until you start the pipeline again. Table states below are from before it stopped.', + showLogsLink: false, + } + } + + if (displayState.key === 'unknown') { + return { + type: 'warning', + title: displayState.title, + description: 'We can’t tell whether replication is running', + showLogsLink: true, + } + } + + const progress = getInitialSyncProgress(tableStatuses) + if (progress.syncingCount === 0) return undefined + + return { + type: 'note', + title: 'Initial sync is running', + description: `${getInitialSyncSummary(progress)} Each table starts streaming as its copy finishes.`, + showLogsLink: false, + } +} diff --git a/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/ReplicationPipelineStatus.test.tsx b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/ReplicationPipelineStatus.test.tsx new file mode 100644 index 00000000000..3006b3ed5fa --- /dev/null +++ b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/ReplicationPipelineStatus.test.tsx @@ -0,0 +1,95 @@ +import { screen } from '@testing-library/react' +import type { components } from 'api-types' +import { HttpResponse } from 'msw' +import { describe, expect, test, vi } from 'vitest' + +import { ReplicationPipelineStatus } from './ReplicationPipelineStatus' +import { PipelineRequestStatusProvider } from '@/state/replication-pipeline-request-status' +import { customRender } from '@/tests/lib/custom-render' +import { addAPIMock } from '@/tests/lib/msw' + +vi.mock('common', async (importOriginal) => ({ + ...(await importOriginal()), + useParams: () => ({ ref: 'default', pipelineId: '42' }), +})) + +type PipelineResponse = components['schemas']['PipelineResponse_Output'] +type PipelineStatusResponse = components['schemas']['PipelineStatusResponse_Output'] +type PipelineReplicationStatusResponse = + components['schemas']['PipelineReplicationStatusResponse_Output'] + +const pipeline: PipelineResponse = { + id: 42, + config: { publication_name: 'analytics_publication' }, + destination_id: 7, + destination_name: 'Analytics warehouse', + replicator_id: 1, + source_id: 2, + source_name: 'main-db', + tenant_id: 'default', +} + +describe('ReplicationPipelineStatus', () => { + test('preserves the overview structure while pipeline details load', async () => { + let resolvePipeline: (value: PipelineResponse) => void = () => {} + let resolvePipelineStatus: (value: PipelineStatusResponse) => void = () => {} + const pipelineResponse = new Promise((resolve) => { + resolvePipeline = resolve + }) + const pipelineStatusResponse = new Promise((resolve) => { + resolvePipelineStatus = resolve + }) + + addAPIMock({ + method: 'get', + path: '/platform/replication/:ref/pipelines/:pipeline_id', + response: async () => HttpResponse.json(await pipelineResponse), + }) + addAPIMock({ + method: 'get', + path: '/platform/replication/:ref/pipelines/:pipeline_id/status', + response: async () => HttpResponse.json(await pipelineStatusResponse), + }) + addAPIMock({ + method: 'get', + path: '/platform/replication/:ref/pipelines/:pipeline_id/replication-status', + response: () => + HttpResponse.json({ + pipeline_id: 42, + apply_lag: { + active: true, + wal_status: 'reserved', + restart_lsn_bytes: 0, + confirmed_flush_lsn_bytes: 0, + safe_wal_size_bytes: null, + }, + table_statuses: [], + }), + }) + + customRender( + + + + ) + + expect(screen.getByRole('status')).toHaveTextContent('Loading pipeline details') + expect(screen.getByRole('heading', { name: 'Pipeline health' })).toBeVisible() + expect(screen.getByRole('heading', { name: 'Replicated tables' })).toBeVisible() + expect(screen.getByRole('columnheader', { name: 'Table' })).toBeVisible() + expect(screen.getByRole('columnheader', { name: 'Status' })).toBeVisible() + expect(screen.getByRole('columnheader', { name: 'Details' })).toBeVisible() + + resolvePipeline(pipeline) + + expect(screen.getByRole('status')).toHaveTextContent('Loading pipeline details') + + resolvePipelineStatus({ + pipeline_id: 42, + status: { name: 'started' }, + }) + + expect(await screen.findByText('No table data yet')).toBeVisible() + expect(screen.getByRole('status')).toHaveTextContent('') + }) +}) diff --git a/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/ReplicationPipelineStatus.tsx b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/ReplicationPipelineStatus.tsx index ab7127b1f9a..8ad3ef6d5aa 100644 --- a/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/ReplicationPipelineStatus.tsx +++ b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/ReplicationPipelineStatus.tsx @@ -1,5 +1,6 @@ import { useParams } from 'common' -import { Activity, ChevronDown, Info, RotateCcw, Search, WifiOff, X } from 'lucide-react' +import { Activity, ChevronDown, RotateCcw, Search, X } from 'lucide-react' +import Link from 'next/link' import { parseAsString, useQueryState } from 'nuqs' import { useMemo, useState } from 'react' import { @@ -17,9 +18,18 @@ import { TableHeadSort, TableRow, } from 'ui' +import { Admonition } from 'ui-patterns/Admonition' import { Input } from 'ui-patterns/DataInputs/Input' +import { EmptyStatePresentational } from 'ui-patterns/EmptyStatePresentational' import { PageContainer } from 'ui-patterns/PageContainer' -import { GenericSkeletonLoader } from 'ui-patterns/ShimmeringLoader' +import { + PageSection, + PageSectionContent, + PageSectionMeta, + PageSectionSummary, + PageSectionTitle, +} from 'ui-patterns/PageSection' +import { GenericTableLoader, ShimmeringLoader } from 'ui-patterns/ShimmeringLoader' import { BatchRestartDialog } from '../BatchRestartDialog' import { ErrorDetailsDialog } from '../ErrorDetailsDialog' @@ -27,6 +37,7 @@ import { getStatusName } from '../Pipeline.utils' import { PipelineStatusName, STATUS_REFRESH_FREQUENCY_MS } from '../Replication.constants' import { RestartTableDialog } from '../RestartTableDialog' import { PipelineHealthSection } from './PipelineHealthSection' +import { getPipelineStateNotice, getTableStatusEmptyState } from './PipelineOverview.utils' import { getDisabledStateConfig } from './ReplicationPipelineStatus.utils' import { TableReplicationRow } from './TableReplicationRow' import { AlertError } from '@/components/ui/AlertError' @@ -59,6 +70,43 @@ const compareTableStates = ( b: ReplicationPipelineTableStatus['state'] ) => TABLE_STATE_SORT_ORDER.indexOf(a.name) - TABLE_STATE_SORT_ORDER.indexOf(b.name) +const PipelineOverviewSkeleton = () => ( + <> + + + + Pipeline health + + + + + +
+ {Array.from({ length: 5 }, (_, index) => ( +
+ + +
+ ))} +
+ + + + + + + + + Replicated tables + + + + + + + +) + /** * Component for displaying replication pipeline status and table replication details. * Supports both legacy 'error' state and new 'errored' state with retry policies. @@ -97,13 +145,14 @@ export const ReplicationPipelineStatus = () => { pipelineId, }) - const { data: pipelineStatusData } = useReplicationPipelineStatusQuery( - { projectRef, pipelineId }, - { - enabled: !!pipelineId, - refetchInterval: STATUS_REFRESH_FREQUENCY_MS, - } - ) + const { data: pipelineStatusData, isPending: isPipelineStatusLoading } = + useReplicationPipelineStatusQuery( + { projectRef, pipelineId }, + { + enabled: !!pipelineId, + refetchInterval: STATUS_REFRESH_FREQUENCY_MS, + } + ) const { data: replicationStatusData, @@ -166,6 +215,7 @@ export const ReplicationPipelineStatus = () => { const hasErroredTables = erroredTables.length > 0 const isAnyRestartInProgress = restartingTableIds.size > 0 + const isLoading = isPipelineLoading || isPipelineStatusLoading || isStatusLoading const hasTableData = tableStatuses.length > 0 const isPipelineActionable = @@ -179,253 +229,278 @@ export const ReplicationPipelineStatus = () => { const isPipelineBusy = isEnablingDisabling || isAnyRestartInProgress const showDisabledState = isPipelineBusy || !isPipelineActionable const canResetErroredTables = hasErroredTables && !isAnyRestartInProgress && !showDisabledState - const lastKnownStateMessage = - statusName === PipelineStatusName.STOPPED - ? 'Showing the last known table state before the pipeline was stopped.' - : statusName === PipelineStatusName.FAILED - ? 'Showing the last reported table state before the pipeline failed.' - : null - const refreshIntervalLabel = - STATUS_REFRESH_FREQUENCY_MS >= 1000 - ? `${Math.round(STATUS_REFRESH_FREQUENCY_MS / 1000)}s` - : `${STATUS_REFRESH_FREQUENCY_MS}ms` + const stateNotice = getPipelineStateNotice({ requestStatus, statusName, tableStatuses }) + const isSlotDisconnected = + !isStatusError && statusName === PipelineStatusName.STARTED && applyLagMetrics?.active === false + const logsUrl = `/project/${projectRef}/logs/replication-logs?f=${encodeURIComponent( + JSON.stringify({ pipeline_id: pipelineId }) + )}` + const emptyState = getTableStatusEmptyState({ + isDisabled: showDisabledState, + disabledStateConfig: config, + statusName, + }) return ( <> - + +

+ {isLoading ? 'Loading pipeline details' : ''} +

+ {isPipelineError && ( - + + + + + )} - {isStatusError && ( -
- - Live updates paused - Retrying automatically -
- )} + {isLoading && } - {(isPipelineLoading || isStatusLoading) && ( -
-
-
-
-
- -
- )} + {!isLoading && ( + + {stateNotice !== undefined && ( + + View logs + + ) : undefined + } + /> + )} + + {hasErroredTables && !showDisabledState && ( + } + disabled={isAnyRestartInProgress || isPipelineError} + loading={isAnyRestartInProgress} + onClick={() => { + setBatchRestartMode('errored') + setShowBatchRestartDialog(true) + }} + > + Reset failed tables + + } + /> + )} + + {isSlotDisconnected && ( + + )} - {applyLagMetrics && ( - {isStatusError && ( -

- Unable to refresh data. Showing the last values we received. -

+ )} )} - {!isPipelineLoading && !isStatusLoading && hasTableData && ( -
-
- } - size="tiny" - className="text-xs w-52" - placeholder="Search tables" - value={searchString} - disabled={isPipelineError} - onChange={(e) => setSearchString(e.target.value)} - onKeyDown={onSearchInputEscape(searchString, setSearchString)} - actions={ - searchString.length > 0 && ( - - - -
-
+
+ + + +
+
- {lastKnownStateMessage !== null && !showDisabledState && ( -
- - {lastKnownStateMessage} -
- )} - - - - - - - - - Table - - - - - Status - - - Details - - - - - - - - {filteredTableStatuses.length === 0 && ( - - )} - {filteredTableStatuses.map((table) => { - const isRestarting = restartingTableIds.has(table.id) - const isErrorState = table.state.name === 'error' - const errorReason = - isErrorState && 'reason' in table.state ? table.state.reason : undefined - const errorSolution = - isErrorState && 'solution' in table.state - ? (table.state.solution ?? undefined) - : undefined - return ( - { - setSelectedTableForRestart({ - id: table.id, - schema: table.schema, - name: table.name, - }) - setShowRestartDialog(true) - }} - onSelectShowError={ - isErrorState && errorReason - ? () => { - setSelectedTableError({ - tableName: `${table.schema}.${table.name}`, - reason: errorReason, - solution: errorSolution, + + +
- {filteredTableStatuses.length === 0 && searchString.length > 0 - ? `No results found for “${searchString}”` - : ''} -
+ + + + + Table + + + + + Status + + + Details + + + + + + + + {filteredTableStatuses.length === 0 && ( + + )} + {filteredTableStatuses.map((table) => { + const isRestarting = restartingTableIds.has(table.id) + const isErrorState = table.state.name === 'error' + const errorReason = + isErrorState && 'reason' in table.state + ? table.state.reason + : undefined + const errorSolution = + isErrorState && 'solution' in table.state + ? (table.state.solution ?? undefined) + : undefined + return ( + { + setSelectedTableForRestart({ + id: table.id, + schema: table.schema, + name: table.name, }) - setShowErrorDialog(true) + setShowRestartDialog(true) + }} + onSelectShowError={ + isErrorState && errorReason + ? () => { + setSelectedTableError({ + tableName: `${table.schema}.${table.name}`, + reason: errorReason, + solution: errorSolution, + }) + setShowErrorDialog(true) + } + : () => {} } - : () => {} - } - /> - ) - })} - -
+ {filteredTableStatuses.length === 0 && searchString.length > 0 + ? `No results found for “${searchString}”` + : ''} +
- - -
- )} - - {!isPipelineLoading && !isStatusLoading && tableStatuses.length === 0 && ( -
-
-
- -
-
-

- {showDisabledState - ? config.title - : statusName === PipelineStatusName.STOPPED - ? 'Pipeline stopped' - : statusName === PipelineStatusName.FAILED - ? 'Pipeline failed' - : 'No table data yet'} -

-

- {showDisabledState - ? config.message - : statusName === PipelineStatusName.STOPPED - ? 'Start the pipeline to begin replication.' - : statusName === PipelineStatusName.FAILED - ? 'The pipeline encountered an error. Restart it or reset your tables to recover.' - : 'Table status will appear here once replication begins.'} -

-
- {statusName !== PipelineStatusName.STOPPED && ( -

- Data refreshes every {refreshIntervalLabel} -

+ /> + ) + })} + + + + +
)} -
- + + {!hasTableData && ( + + )} + + )} diff --git a/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/TableReplicationRow.tsx b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/TableReplicationRow.tsx index 4b5b0c970cc..836efe04319 100644 --- a/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/TableReplicationRow.tsx +++ b/apps/studio/components/interfaces/Database/Replication/ReplicationPipelineStatus/TableReplicationRow.tsx @@ -115,7 +115,7 @@ export const TableReplicationRow = ({ icon={} /> - + - +