Files
Riccardo Busetti edec85d1ca fix(pipelines): Make pipeline actions and status updates reliable (#50085)
## Summary

Make pipeline actions and status feedback reliable while requests are
running or fail. Let the backend coordinate table resets and restarts,
keep stopped pipelines stopped after resets or settings changes, and
refresh the UI from confirmed backend state.

## Pipeline actions and recovery

- Reset one table, all errored tables, or all tables through the
rollback endpoint without separate frontend stop/start requests. Explain
which destination data is deleted, which rows are copied again, initial
sync charges, and the skip-initial-sync setting.
- Keep pending feedback until the action and a fresh status read finish,
including across navigation and polling errors. Prevent overlapping
actions and disable start/stop controls when status is unavailable or
transitioning.
- Close the creation form once the pipeline is created. If its initial
start fails, users can retry Start on the existing pipeline without
creating a duplicate.
- Wait for confirmed shutdown before deletion; a shutdown error or
timeout leaves deletion retryable. Keep failed version updates open and
avoid reporting success.
- Clarify recovery guidance and pending labels, suppress duplicate error
toasts, and hide stale table errors during transitions.

## Status updates and shared UI

- Poll pipeline status and table metrics one second after each response,
share in-flight reads, pause dashboard polling in background tabs, and
respect rate-limit backoff. The shutdown waiter continues in the
background.
- Refresh metadata after mutations even when an older read is in flight,
while preserving shared polling requests. Refresh affected data after
failures that may follow a committed reset or settings change.
- Move pending request state into the shared, project-keyed
`DatabaseLayout` so the list, detail page, and diagram stay consistent.
The surrounding database-page changes update named imports in both
Next.js and TanStack routes.
- Simplify action, status, and form rendering; announce status changes
to assistive technology; and sort table statuses without mutating cached
data.

---------

Co-authored-by: Joshen Lim <joshenlimek@gmail.com>
Co-authored-by: Danny White <3104761+dnywh@users.noreply.github.com>
2026-09-18 11:32:48 +08:00

282 lines
11 KiB
TypeScript

import { useParams } from 'common'
import { ChevronRight, Minus } from 'lucide-react'
import { useRouter } from 'next/router'
import { useState } from 'react'
import { toast } from 'sonner'
import { TableCell, TableRow } from 'ui'
import { ShimmeringLoader } from 'ui-patterns/ShimmeringLoader'
import { DeleteDestination } from './DeleteDestination'
import { DestinationLogo } from './DestinationLogo'
import { DetailSubtext } from './DetailSubtext'
import { PipelineStatePill } from './PipelineStatePill'
import { PipelineStatusName } from './Replication.constants'
import {
getFormattedLagValue,
getInitialSyncProgress,
} from './ReplicationPipelineStatus/ReplicationPipelineStatus.utils'
import { RowMenu } from './RowMenu'
import { UpdateVersionModal } from './UpdateVersionModal'
import { useDestinationInformation } from './useDestinationInformation'
import { AlertError } from '@/components/ui/AlertError'
import { useDeleteDestinationPipelineMutation } from '@/data/replication/delete-destination-pipeline-mutation'
import { useReplicationPipelineReplicationStatusQuery } from '@/data/replication/pipeline-replication-status-query'
import { useReplicationPipelineStatusQuery } from '@/data/replication/pipeline-status-query'
import { useReplicationPipelineVersionQuery } from '@/data/replication/pipeline-version-query'
import { useStopPipelineMutation } from '@/data/replication/stop-pipeline-mutation'
import { createNavigationHandler } from '@/lib/navigation'
import {
PipelineStatusRequestStatus,
usePipelineRequestStatus,
} from '@/state/replication-pipeline-request-status'
import { type ResponseError } from '@/types'
interface DestinationRowProps {
destinationId: number
}
export const DestinationRow = ({ destinationId }: DestinationRowProps) => {
const router = useRouter()
const { ref: projectRef } = useParams()
const [showDeleteDestinationForm, setShowDeleteDestinationForm] = useState(false)
const [isDeleting, setIsDeleting] = useState(false)
const [showUpdateVersionModal, setShowUpdateVersionModal] = useState(false)
const { type, statusName, destination, pipeline, pipelineStatus, pipelineFetcher } =
useDestinationInformation({
id: destinationId,
})
const {
error: pipelineError,
isPending: isPipelineLoading,
isError: isPipelineError,
isSuccess: isPipelineSuccess,
} = pipelineFetcher
const destinationName = destination?.name ?? ''
const {
error: pipelineStatusError,
isPending: isPipelineStatusLoading,
isError: isPipelineStatusError,
isSuccess: isPipelineStatusSuccess,
} = useReplicationPipelineStatusQuery({
projectRef,
pipelineId: pipeline?.id,
})
const { getRequestStatus } = usePipelineRequestStatus()
const requestStatus = pipeline?.id
? getRequestStatus(pipeline.id)
: PipelineStatusRequestStatus.None
const { mutateAsync: stopPipeline } = useStopPipelineMutation({ onError: () => {} })
const { mutateAsync: deleteDestinationPipeline } = useDeleteDestinationPipelineMutation({
onError: () => {},
})
// Fetch table-level replication status to surface errors in list view
const {
data: replicationStatusData,
isPending: isReplicationStatusLoading,
isError: isReplicationStatusError,
} = useReplicationPipelineReplicationStatusQuery({ projectRef, pipelineId: pipeline?.id }, {})
const tableStatuses = replicationStatusData?.table_statuses ?? []
const errorCount = tableStatuses.filter((t) => t.state?.name === 'error').length
const applyLag = replicationStatusData?.apply_lag
// Show the byte-based slot lag (WAL the destination hasn't confirmed flushing yet). The
// time-based flush_lag from pg_stat_replication is routinely NULL for logical slots that are
// idle or don't report timed feedback, whereas confirmed_flush_lsn_bytes is always populated.
const lagBytes = applyLag?.confirmed_flush_lsn_bytes
const lag = getFormattedLagValue('bytes', lagBytes)
// The lag figure only covers ongoing changes, so it reads as "Caught up" while an initial copy
// is still running. Say what's actually happening instead.
const { syncingCount } = getInitialSyncProgress(tableStatuses)
const isInitialSyncRunning = syncingCount > 0
const isCaughtUp = lagBytes === 0
// Hide old table errors while an optimistic lifecycle action is displayed.
const isPipelineStopped = statusName === PipelineStatusName.STOPPED
const isTransitioning = requestStatus !== PipelineStatusRequestStatus.None
const hasTableErrors = errorCount > 0 && !isPipelineStopped && !isTransitioning
// Check if a newer pipeline version is available (one-time check cached for session)
const { data: versionData } = useReplicationPipelineVersionQuery({
projectRef,
pipelineId: pipeline?.id,
})
const hasUpdate = Boolean(versionData?.new_version)
const handleNavigation = pipeline
? createNavigationHandler(`/project/${projectRef}/database/replication/${pipeline.id}`, router)
: undefined
const onDeleteClick = async () => {
if (!projectRef) {
return console.error('Project ref is required')
}
if (!pipeline) {
return toast.error('No pipeline found')
}
try {
setIsDeleting(true)
await stopPipeline({ projectRef, pipelineId: pipeline.id, waitUntilStopped: true })
await deleteDestinationPipeline({
projectRef,
destinationId: destinationId,
pipelineId: pipeline.id,
})
// Close dialog after successful deletion
setShowDeleteDestinationForm(false)
toast.success(`Deleted pipeline "${destinationName}"`)
} catch (error) {
toast.error(`Failed to delete pipeline: ${(error as ResponseError).message}`)
} finally {
setIsDeleting(false)
}
}
// Five distinct states, so early returns rather than a ternary chain. The row only renders once
// a pipeline exists, so there is no "no pipeline" case to handle here.
const renderLag = () => {
if (isReplicationStatusLoading) return <ShimmeringLoader />
if (isInitialSyncRunning)
return <span className="text-foreground-light whitespace-nowrap">Initial sync</span>
if (isReplicationStatusError || !applyLag)
return (
<>
<Minus size={18} className="text-foreground-lighter" aria-hidden />
<span className="sr-only">Lag unavailable</span>
</>
)
if (isCaughtUp)
return <span className="text-foreground-light whitespace-nowrap">Caught up</span>
return <span className="text-foreground-light whitespace-nowrap">{lag.display}</span>
}
return (
<>
{isPipelineError && (
<TableRow>
<TableCell colSpan={6}>
<AlertError error={pipelineError} subject="Failed to retrieve pipeline information" />
</TableCell>
</TableRow>
)}
{isPipelineSuccess && pipeline && (
<TableRow
className="relative cursor-pointer focus-inset"
onClick={handleNavigation}
onAuxClick={handleNavigation}
onKeyDown={handleNavigation}
tabIndex={0}
>
<TableCell className="!pr-1">
{type ? <DestinationLogo type={type} hasErrors={hasTableErrors} /> : null}
</TableCell>
<TableCell className="max-w-[180px]">
{isPipelineLoading ? (
<ShimmeringLoader />
) : (
<div className="flex flex-col gap-y-0.5">
<p className="text-sm font-medium text-foreground truncate">
{destinationName || type}
</p>
<DetailSubtext className="flex items-center gap-x-1.5">
<span>#{pipeline?.id}</span>
<span aria-hidden>&middot;</span>
<span>{type}</span>
{hasTableErrors && (
<>
<span aria-hidden>&middot;</span>
<span className="text-destructive">
{errorCount} table error{errorCount === 1 ? '' : 's'}
</span>
</>
)}
</DetailSubtext>
</div>
)}
</TableCell>
<TableCell>
{isPipelineLoading || !pipeline ? (
<ShimmeringLoader />
) : (
<PipelineStatePill
pipelineStatus={pipelineStatus?.status}
error={pipelineStatusError}
isLoading={isPipelineStatusLoading}
isError={isPipelineStatusError}
isSuccess={isPipelineStatusSuccess}
requestStatus={requestStatus}
projectRef={projectRef}
pipelineId={pipeline?.id}
/>
)}
</TableCell>
<TableCell>
<div aria-live="polite" aria-atomic="true">
{isReplicationStatusLoading && <span className="sr-only">Loading lag</span>}
{renderLag()}
</div>
</TableCell>
<TableCell>
{isPipelineLoading || !pipeline ? (
<ShimmeringLoader />
) : (
pipeline.config.publication_name
)}
</TableCell>
<TableCell>
<div className="flex items-center justify-end gap-x-2">
<div
onClick={(event) => event.stopPropagation()}
onAuxClick={(event) => event.stopPropagation()}
onKeyDown={(event) => event.stopPropagation()}
>
<RowMenu
destinationId={destinationId}
pipeline={pipeline}
pipelineStatus={pipelineStatus?.status}
error={pipelineStatusError}
isLoading={isPipelineStatusLoading}
isError={isPipelineStatusError}
onDeleteClick={() => setShowDeleteDestinationForm(true)}
hasUpdate={hasUpdate}
onUpdateClick={() => setShowUpdateVersionModal(true)}
/>
</div>
<ChevronRight
size={16}
strokeWidth={1.5}
className="text-foreground-lighter"
aria-hidden
/>
<button tabIndex={-1} className="sr-only">
Go to pipeline details
</button>
</div>
</TableCell>
</TableRow>
)}
<DeleteDestination
visible={showDeleteDestinationForm}
setVisible={setShowDeleteDestinationForm}
onDelete={onDeleteClick}
isLoading={isDeleting}
name={destinationName}
/>
<UpdateVersionModal
visible={showUpdateVersionModal}
pipeline={pipeline}
onClose={() => setShowUpdateVersionModal(false)}
/>
</>
)
}