mirror of
https://github.com/Syngnat/GoNavi.git
synced 2026-08-25 18:21:22 +08:00
Compare commits
1 Commits
v0.9.4
...
codex/issu
| Author | SHA1 | Date | |
|---|---|---|---|
|
88841af92d |
@@ -0,0 +1,113 @@
|
||||
import React from 'react';
|
||||
import TestRenderer, { act } from 'react-test-renderer';
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
|
||||
import { DataSyncScheduleView } from './DataSyncOperationalViews';
|
||||
import type { DataSyncScheduleSummary } from './model';
|
||||
import { createDataSyncWorkbenchTranslate } from './text';
|
||||
|
||||
const schedule: DataSyncScheduleSummary = {
|
||||
id: 'orders:schedule',
|
||||
taskId: 'orders',
|
||||
taskName: 'Orders to warehouse',
|
||||
revision: 7,
|
||||
lifecycle: 'enabled',
|
||||
enabled: true,
|
||||
expression: '0 */5 * * * *',
|
||||
timezone: 'Asia/Shanghai',
|
||||
nextRunAt: '2026-08-21T03:05:00.000Z',
|
||||
latestRun: {
|
||||
id: 'orders:run:failed',
|
||||
status: 'failed',
|
||||
startedAt: '2026-08-21T03:00:00.000Z',
|
||||
finishedAt: '2026-08-21T03:00:12.000Z',
|
||||
errorSummary: 'target write failed: password=[REDACTED]',
|
||||
},
|
||||
};
|
||||
|
||||
describe('DataSyncScheduleView', () => {
|
||||
it('shows the latest failure and routes each control through its supplied callback', () => {
|
||||
const onRefresh = vi.fn();
|
||||
const onToggle = vi.fn();
|
||||
const onRunNow = vi.fn();
|
||||
const onViewRun = vi.fn();
|
||||
const renderer = TestRenderer.create(
|
||||
<DataSyncScheduleView
|
||||
schedules={[schedule]}
|
||||
t={createDataSyncWorkbenchTranslate('en-US')}
|
||||
refreshing={false}
|
||||
onRefresh={onRefresh}
|
||||
onToggle={onToggle}
|
||||
onRunNow={onRunNow}
|
||||
onViewRun={onViewRun}
|
||||
/>,
|
||||
);
|
||||
|
||||
expect(
|
||||
renderer.root.findByProps({ 'data-task-id': schedule.taskId }).props['data-enabled'],
|
||||
).toBe('true');
|
||||
expect(renderer.root.findAllByProps({ children: 'Failed' })).toHaveLength(1);
|
||||
expect(
|
||||
renderer.root
|
||||
.findAllByType('small')
|
||||
.some((node) =>
|
||||
node.children.join('') ===
|
||||
`${schedule.latestRun!.startedAt} \u2192 ${schedule.latestRun!.finishedAt}`,
|
||||
),
|
||||
).toBe(true);
|
||||
expect(renderer.root.findAllByProps({ children: schedule.latestRun!.errorSummary }))
|
||||
.toHaveLength(1);
|
||||
|
||||
const button = (label: string) =>
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((candidate) => candidate.children.includes(label))!;
|
||||
act(() => {
|
||||
button('Refresh').props.onClick();
|
||||
button('Disable').props.onClick();
|
||||
button('Run now').props.onClick();
|
||||
button('View run').props.onClick();
|
||||
});
|
||||
|
||||
expect(onRefresh).toHaveBeenCalledTimes(1);
|
||||
expect(onToggle).toHaveBeenCalledWith(schedule);
|
||||
expect(onRunNow).toHaveBeenCalledWith(schedule);
|
||||
expect(onViewRun).toHaveBeenCalledWith(schedule.latestRun!.id);
|
||||
});
|
||||
|
||||
it('disables both schedule mutations while one is in flight', () => {
|
||||
const renderer = TestRenderer.create(
|
||||
<DataSyncScheduleView
|
||||
schedules={[schedule]}
|
||||
t={createDataSyncWorkbenchTranslate('en-US')}
|
||||
refreshing={false}
|
||||
onRefresh={vi.fn()}
|
||||
onToggle={vi.fn()}
|
||||
onRunNow={vi.fn()}
|
||||
busyAction={`disable:${schedule.taskId}`}
|
||||
/>,
|
||||
);
|
||||
const button = (label: string) =>
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((candidate) => candidate.children.includes(label))!;
|
||||
|
||||
expect(button('Disable').props.disabled).toBe(true);
|
||||
expect(button('Run now').props.disabled).toBe(true);
|
||||
|
||||
renderer.update(
|
||||
<DataSyncScheduleView
|
||||
schedules={[schedule]}
|
||||
t={createDataSyncWorkbenchTranslate('en-US')}
|
||||
refreshing={false}
|
||||
onRefresh={vi.fn()}
|
||||
onToggle={vi.fn()}
|
||||
onRunNow={vi.fn()}
|
||||
busyAction={`run-now:${schedule.taskId}`}
|
||||
/>,
|
||||
);
|
||||
|
||||
expect(button('Disable').props.disabled).toBe(true);
|
||||
expect(button('Run now').props.disabled).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -260,13 +260,31 @@ export const DataSyncScheduleView: React.FC<{
|
||||
t: DataSyncWorkbenchTranslate;
|
||||
refreshing: boolean;
|
||||
onRefresh: () => void;
|
||||
}> = ({ schedules, t, refreshing, onRefresh }) => (
|
||||
busyAction?: string;
|
||||
onToggle?: (schedule: DataSyncScheduleSummary) => void;
|
||||
onRunNow?: (schedule: DataSyncScheduleSummary) => void;
|
||||
onViewRun?: (runId: string) => void;
|
||||
}> = ({
|
||||
schedules,
|
||||
t,
|
||||
refreshing,
|
||||
onRefresh,
|
||||
busyAction = '',
|
||||
onToggle,
|
||||
onRunNow,
|
||||
onViewRun,
|
||||
}) => (
|
||||
<section className="gn-data-sync-operational-view" data-data-sync-schedules="true">
|
||||
<header className="gn-data-sync-view-heading">
|
||||
<h1>{t('schedules.title')}</h1>
|
||||
<div>
|
||||
<p>{t('schedules.subtitle')}</p>
|
||||
<button type="button" className="gn-data-sync-button" disabled={refreshing} onClick={onRefresh}>
|
||||
<button
|
||||
type="button"
|
||||
className="gn-data-sync-button"
|
||||
disabled={refreshing || Boolean(busyAction)}
|
||||
onClick={onRefresh}
|
||||
>
|
||||
{t('common.refresh')}
|
||||
</button>
|
||||
</div>
|
||||
@@ -277,17 +295,104 @@ export const DataSyncScheduleView: React.FC<{
|
||||
description={t('schedules.empty_desc')}
|
||||
/>
|
||||
) : (
|
||||
<ul className="gn-data-sync-summary-list">
|
||||
{schedules.map((schedule) => (
|
||||
<li key={schedule.id}>
|
||||
<span className="gn-data-sync-summary-list__signal" data-active={schedule.enabled} />
|
||||
<strong>{schedule.taskName}</strong>
|
||||
<code>{schedule.expression}</code>
|
||||
<span>{schedule.timezone}</span>
|
||||
<time>{schedule.nextRunAt}</time>
|
||||
</li>
|
||||
))}
|
||||
</ul>
|
||||
<div className="gn-data-sync-schedule-table" role="table">
|
||||
<div className="gn-data-sync-schedule-table__head" role="row">
|
||||
<span>{t('schedules.task')}</span>
|
||||
<span>{t('schedules.status')}</span>
|
||||
<span>{t('schedules.trigger')}</span>
|
||||
<span>{t('schedules.next_run')}</span>
|
||||
<span>{t('schedules.latest_run')}</span>
|
||||
<span>{t('schedules.actions')}</span>
|
||||
</div>
|
||||
{schedules.map((schedule) => {
|
||||
const latest = schedule.latestRun;
|
||||
const scheduleActionInFlight = Boolean(busyAction);
|
||||
return (
|
||||
<div
|
||||
key={schedule.id}
|
||||
className="gn-data-sync-schedule-table__row"
|
||||
role="row"
|
||||
data-task-id={schedule.taskId}
|
||||
data-enabled={schedule.enabled ? 'true' : 'false'}
|
||||
>
|
||||
<div className="gn-data-sync-schedule-table__task" role="cell">
|
||||
<span
|
||||
className="gn-data-sync-summary-list__signal"
|
||||
data-active={schedule.enabled}
|
||||
aria-hidden="true"
|
||||
/>
|
||||
<strong>{schedule.taskName}</strong>
|
||||
{schedule.lifecycle ? (
|
||||
<small>{t(`task_list.lifecycle.${schedule.lifecycle}`)}</small>
|
||||
) : null}
|
||||
</div>
|
||||
<span role="cell">
|
||||
{schedule.enabled
|
||||
? t('schedules.enabled')
|
||||
: t('schedules.disabled')}
|
||||
</span>
|
||||
<span role="cell">
|
||||
<code>{schedule.expression}</code>
|
||||
<small>{schedule.timezone}</small>
|
||||
</span>
|
||||
<time role="cell">{schedule.nextRunAt || '—'}</time>
|
||||
<div className="gn-data-sync-schedule-table__latest" role="cell">
|
||||
{latest ? (
|
||||
<>
|
||||
<span className="gn-data-sync-state-label" data-state={latest.status}>
|
||||
{t(`status.${latest.status}`)}
|
||||
</span>
|
||||
<small>
|
||||
{latest.startedAt || '—'}
|
||||
{latest.finishedAt ? ` → ${latest.finishedAt}` : ''}
|
||||
</small>
|
||||
{latest.errorSummary ? (
|
||||
<span className="gn-data-sync-schedule-table__error">
|
||||
{latest.errorSummary}
|
||||
</span>
|
||||
) : null}
|
||||
{onViewRun ? (
|
||||
<button
|
||||
type="button"
|
||||
className="gn-data-sync-link-button"
|
||||
onClick={() => onViewRun(latest.id)}
|
||||
>
|
||||
{t('schedules.view_run')}
|
||||
</button>
|
||||
) : null}
|
||||
</>
|
||||
) : (
|
||||
<span>{t('schedules.no_runs')}</span>
|
||||
)}
|
||||
</div>
|
||||
<div className="gn-data-sync-schedule-table__actions" role="cell">
|
||||
{onToggle ? (
|
||||
<button
|
||||
type="button"
|
||||
className="gn-data-sync-button"
|
||||
disabled={scheduleActionInFlight}
|
||||
onClick={() => onToggle(schedule)}
|
||||
>
|
||||
{schedule.enabled
|
||||
? t('schedules.disable')
|
||||
: t('schedules.enable')}
|
||||
</button>
|
||||
) : null}
|
||||
{onRunNow ? (
|
||||
<button
|
||||
type="button"
|
||||
className="gn-data-sync-button gn-data-sync-button--primary"
|
||||
disabled={scheduleActionInFlight}
|
||||
onClick={() => onRunNow(schedule)}
|
||||
>
|
||||
{t('schedules.run_now')}
|
||||
</button>
|
||||
) : null}
|
||||
</div>
|
||||
</div>
|
||||
);
|
||||
})}
|
||||
</div>
|
||||
)}
|
||||
</section>
|
||||
);
|
||||
|
||||
@@ -5,16 +5,15 @@ import { afterEach, describe, expect, it, vi } from 'vitest';
|
||||
const runtimeApi = vi.hoisted(() => ({
|
||||
ClipboardSetText: vi.fn(() => Promise.resolve(false)),
|
||||
}));
|
||||
const clipboardWriteText = vi.hoisted(() => vi.fn(() => Promise.resolve()));
|
||||
|
||||
vi.mock('../../../wailsjs/runtime/runtime', () => runtimeApi);
|
||||
|
||||
import { DataSyncPreflightPanel } from './DataSyncPreflightPanel';
|
||||
|
||||
Object.assign(globalThis, {
|
||||
navigator: {
|
||||
clipboard: {
|
||||
writeText: vi.fn(() => Promise.resolve()),
|
||||
},
|
||||
vi.stubGlobal('navigator', {
|
||||
clipboard: {
|
||||
writeText: clipboardWriteText,
|
||||
},
|
||||
});
|
||||
import {
|
||||
@@ -58,8 +57,8 @@ afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
runtimeApi.ClipboardSetText.mockReset();
|
||||
runtimeApi.ClipboardSetText.mockResolvedValue(false);
|
||||
(navigator.clipboard.writeText as ReturnType<typeof vi.fn>).mockReset();
|
||||
(navigator.clipboard.writeText as ReturnType<typeof vi.fn>).mockResolvedValue(undefined);
|
||||
clipboardWriteText.mockReset();
|
||||
clipboardWriteText.mockResolvedValue(undefined);
|
||||
});
|
||||
|
||||
describe('DataSyncPreflightPanel production approval', () => {
|
||||
@@ -232,7 +231,9 @@ describe('DataSyncPreflightPanel production approval', () => {
|
||||
t={t}
|
||||
onLocateIssue={() => undefined}
|
||||
approvalChallenge={{
|
||||
taskId: 'task-1',
|
||||
definitionHash: 'hash-1',
|
||||
taskRevision: 4,
|
||||
notBefore: new Date(now + 10_000).toISOString(),
|
||||
expiresAt: new Date(now + 120_000).toISOString(),
|
||||
}}
|
||||
|
||||
@@ -82,13 +82,17 @@ export const DataSyncPreflightPanel: React.FC<{
|
||||
const approvalCurrent = Boolean(
|
||||
snapshot &&
|
||||
approval &&
|
||||
approval.taskId === snapshot.taskId &&
|
||||
approval.definitionHash === snapshot.definitionHash &&
|
||||
approval.taskRevision === snapshot.taskRevision &&
|
||||
Date.parse(approval.expiresAt) > clock,
|
||||
);
|
||||
const challengeCurrent = Boolean(
|
||||
snapshot &&
|
||||
approvalChallenge &&
|
||||
approvalChallenge.taskId === snapshot.taskId &&
|
||||
approvalChallenge.definitionHash === snapshot.definitionHash &&
|
||||
approvalChallenge.taskRevision === snapshot.taskRevision &&
|
||||
Date.parse(approvalChallenge.expiresAt) > clock,
|
||||
);
|
||||
const remainingSeconds = useMemo(() => {
|
||||
|
||||
@@ -1872,6 +1872,83 @@ textarea.gn-data-sync-control {
|
||||
background: var(--gn-danger, #dc2626);
|
||||
}
|
||||
|
||||
.gn-data-sync-schedule-table {
|
||||
min-width: 980px;
|
||||
border-top: 0.5px solid var(--gn-br-2, rgba(15, 23, 42, 0.12));
|
||||
}
|
||||
|
||||
.gn-data-sync-schedule-table__head,
|
||||
.gn-data-sync-schedule-table__row {
|
||||
display: grid;
|
||||
grid-template-columns: minmax(170px, 1.2fr) minmax(90px, 0.65fr) minmax(130px, 0.9fr) minmax(150px, 1fr) minmax(250px, 1.7fr) minmax(190px, 1fr);
|
||||
align-items: center;
|
||||
gap: 12px;
|
||||
padding: 8px 10px;
|
||||
}
|
||||
|
||||
.gn-data-sync-schedule-table__head {
|
||||
color: var(--gn-fg-5, #9ca3af);
|
||||
font-size: var(--gn-font-size-xs, 11px);
|
||||
font-weight: 650;
|
||||
text-transform: uppercase;
|
||||
}
|
||||
|
||||
.gn-data-sync-schedule-table__row {
|
||||
min-height: 62px;
|
||||
border-bottom: 0.5px solid var(--gn-br-2, rgba(15, 23, 42, 0.12));
|
||||
color: var(--gn-fg-2, #1f2937);
|
||||
font-size: var(--gn-font-size-sm, 12px);
|
||||
}
|
||||
|
||||
.gn-data-sync-schedule-table__row > * {
|
||||
min-width: 0;
|
||||
}
|
||||
|
||||
.gn-data-sync-schedule-table__task,
|
||||
.gn-data-sync-schedule-table__latest,
|
||||
.gn-data-sync-schedule-table__actions {
|
||||
display: flex;
|
||||
min-width: 0;
|
||||
align-items: center;
|
||||
gap: 7px;
|
||||
}
|
||||
|
||||
.gn-data-sync-schedule-table__task {
|
||||
flex-wrap: wrap;
|
||||
}
|
||||
|
||||
.gn-data-sync-schedule-table__task strong {
|
||||
overflow: hidden;
|
||||
text-overflow: ellipsis;
|
||||
white-space: nowrap;
|
||||
}
|
||||
|
||||
.gn-data-sync-schedule-table__task small,
|
||||
.gn-data-sync-schedule-table__latest small,
|
||||
.gn-data-sync-schedule-table__row > span small {
|
||||
display: block;
|
||||
width: 100%;
|
||||
color: var(--gn-fg-4, #6b7280);
|
||||
font-size: var(--gn-font-size-xs, 11px);
|
||||
}
|
||||
|
||||
.gn-data-sync-schedule-table__latest {
|
||||
flex-wrap: wrap;
|
||||
align-content: center;
|
||||
}
|
||||
|
||||
.gn-data-sync-schedule-table__error {
|
||||
width: 100%;
|
||||
overflow: hidden;
|
||||
color: var(--gn-danger, #dc2626);
|
||||
text-overflow: ellipsis;
|
||||
white-space: nowrap;
|
||||
}
|
||||
|
||||
.gn-data-sync-schedule-table__actions {
|
||||
flex-wrap: wrap;
|
||||
}
|
||||
|
||||
.gn-data-sync-visually-hidden {
|
||||
position: absolute !important;
|
||||
width: 1px !important;
|
||||
|
||||
@@ -3,7 +3,10 @@ import { renderToStaticMarkup } from 'react-dom/server';
|
||||
import TestRenderer, { act } from 'react-test-renderer';
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest';
|
||||
|
||||
import { createStaticDataSyncWorkbenchGateway } from './gateway';
|
||||
import {
|
||||
createStaticDataSyncWorkbenchGateway,
|
||||
type StaticDataSyncGatewayFixtures,
|
||||
} from './gateway';
|
||||
import {
|
||||
createDataSyncTableMapping,
|
||||
createDataSyncTaskDraft,
|
||||
@@ -301,6 +304,61 @@ describe('DataSyncWorkbenchShell', () => {
|
||||
expect(renderer.root.findByProps({ 'data-dirty': 'false' })).toBeTruthy();
|
||||
});
|
||||
|
||||
it('refreshes the schedule list after saving a newly persisted scheduled task', async () => {
|
||||
const task = {
|
||||
...buildTask(),
|
||||
id: 'data-sync-local-scheduled',
|
||||
lifecycle: 'paused' as const,
|
||||
trigger: {
|
||||
mode: 'interval' as const,
|
||||
intervalSeconds: 300,
|
||||
timezone: 'Asia/Shanghai',
|
||||
},
|
||||
};
|
||||
let persistedTasks: typeof task[] = [];
|
||||
const baseGateway = createStaticDataSyncWorkbenchGateway({ tasks: [] });
|
||||
const listTasks = vi.fn(async () => persistedTasks);
|
||||
const saveTask = vi.fn(async (submitted: typeof task) => {
|
||||
const saved = {
|
||||
...submitted,
|
||||
id: 'persisted-scheduled-task',
|
||||
revision: submitted.revision + 1,
|
||||
};
|
||||
persistedTasks = [saved];
|
||||
return saved;
|
||||
});
|
||||
const gateway = { ...baseGateway, listTasks, saveTask };
|
||||
const renderer = TestRenderer.create(
|
||||
<DataSyncWorkbenchShell initialTasks={[task]} gateway={gateway} locale="en-US" />,
|
||||
);
|
||||
|
||||
await act(async () => {
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
await act(async () => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Save draft'))!
|
||||
.props.onClick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(listTasks.mock.calls.length).toBeGreaterThan(1);
|
||||
act(() => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Schedules'))!
|
||||
.props.onClick();
|
||||
});
|
||||
expect(
|
||||
renderer.root.findByProps({ 'data-task-id': 'persisted-scheduled-task' }),
|
||||
).toBeTruthy();
|
||||
});
|
||||
|
||||
it('shows approval-required preflight state and keeps execution fail-closed', async () => {
|
||||
const task = buildTask();
|
||||
const gateway = createStaticDataSyncWorkbenchGateway({
|
||||
@@ -409,7 +467,7 @@ describe('DataSyncWorkbenchShell', () => {
|
||||
});
|
||||
const resetCheckpoint = vi.fn(baseGateway.resetCheckpoint.bind(baseGateway));
|
||||
const gateway = { ...baseGateway, resetCheckpoint };
|
||||
const confirm = vi.fn(() => false);
|
||||
const confirm = vi.fn<(message?: string) => boolean>(() => false);
|
||||
vi.stubGlobal('confirm', confirm);
|
||||
const renderer = TestRenderer.create(
|
||||
<DataSyncWorkbenchShell initialTasks={[task]} gateway={gateway} locale="en-US" />,
|
||||
@@ -455,4 +513,620 @@ describe('DataSyncWorkbenchShell', () => {
|
||||
expect(resetCheckpoint).toHaveBeenCalledWith(task.id, task.revision);
|
||||
expect(resetButton().props.disabled).toBe(true);
|
||||
});
|
||||
|
||||
it('confirms schedule pause scope, preserves the stored revision, and refreshes the row', async () => {
|
||||
const task = {
|
||||
...buildTask(),
|
||||
lifecycle: 'enabled' as const,
|
||||
trigger: {
|
||||
mode: 'interval' as const,
|
||||
intervalSeconds: 300,
|
||||
timezone: 'Asia/Shanghai',
|
||||
},
|
||||
};
|
||||
const baseGateway = createStaticDataSyncWorkbenchGateway({ tasks: [task] });
|
||||
const saveTask = vi.fn(baseGateway.saveTask.bind(baseGateway));
|
||||
const listTasks = vi.fn(baseGateway.listTasks.bind(baseGateway));
|
||||
const gateway = { ...baseGateway, listTasks, saveTask };
|
||||
const confirm = vi.fn<(message?: string) => boolean>(() => false);
|
||||
vi.stubGlobal('confirm', confirm);
|
||||
const renderer = TestRenderer.create(
|
||||
<DataSyncWorkbenchShell initialTasks={[task]} gateway={gateway} locale="en-US" />,
|
||||
);
|
||||
|
||||
await act(async () => {
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
act(() => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Schedules'))!
|
||||
.props.onClick();
|
||||
});
|
||||
const disable = () =>
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Disable'))!;
|
||||
|
||||
await act(async () => {
|
||||
disable().props.onClick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(confirm).toHaveBeenCalledTimes(1);
|
||||
const confirmationMessage = confirm.mock.calls[0]?.[0] || '';
|
||||
expect(confirmationMessage).toContain(task.name);
|
||||
expect(confirmationMessage).toContain('Source: MySQL 生产库 / sales');
|
||||
expect(confirmationMessage).toContain('Target: PostgreSQL 数仓 / warehouse / ods');
|
||||
expect(saveTask).not.toHaveBeenCalled();
|
||||
|
||||
confirm.mockReturnValue(true);
|
||||
await act(async () => {
|
||||
disable().props.onClick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(listTasks.mock.calls.length).toBeGreaterThan(1);
|
||||
expect(saveTask).toHaveBeenCalledWith(expect.objectContaining({
|
||||
id: task.id,
|
||||
lifecycle: 'paused',
|
||||
revision: task.revision,
|
||||
}));
|
||||
expect(
|
||||
renderer.root.findByProps({ 'data-task-id': task.id }).props['data-enabled'],
|
||||
).toBe('false');
|
||||
});
|
||||
|
||||
it('runs an eligible schedule through current preflight and opens the queued run', async () => {
|
||||
const task = {
|
||||
...buildTask(),
|
||||
lifecycle: 'enabled' as const,
|
||||
trigger: {
|
||||
mode: 'cron' as const,
|
||||
expression: '0 */5 * * * *',
|
||||
timezone: 'Asia/Shanghai',
|
||||
overlap: 'skip' as const,
|
||||
},
|
||||
};
|
||||
const baseGateway = createStaticDataSyncWorkbenchGateway({
|
||||
tasks: [task],
|
||||
capabilities: {
|
||||
[task.id]: {
|
||||
level: 'full',
|
||||
canExecute: true,
|
||||
supportsAutoCreate: true,
|
||||
supportsCdc: false,
|
||||
},
|
||||
},
|
||||
});
|
||||
const preflightTask = vi.fn(baseGateway.preflightTask.bind(baseGateway));
|
||||
const resolveCapability = vi.fn(baseGateway.resolveCapability.bind(baseGateway));
|
||||
const startTask = vi.fn(baseGateway.startTask.bind(baseGateway));
|
||||
const gateway = {
|
||||
...baseGateway,
|
||||
preflightTask,
|
||||
resolveCapability,
|
||||
startTask,
|
||||
};
|
||||
const confirm = vi.fn<(message?: string) => boolean>(() => true);
|
||||
vi.stubGlobal('confirm', confirm);
|
||||
const renderer = TestRenderer.create(
|
||||
<DataSyncWorkbenchShell initialTasks={[task]} gateway={gateway} locale="en-US" />,
|
||||
);
|
||||
|
||||
await act(async () => {
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
act(() => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Schedules'))!
|
||||
.props.onClick();
|
||||
});
|
||||
await act(async () => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Run now'))!
|
||||
.props.onClick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(preflightTask).toHaveBeenCalledWith(expect.objectContaining({
|
||||
id: task.id,
|
||||
revision: task.revision,
|
||||
}));
|
||||
expect(confirm).toHaveBeenCalledWith(expect.stringContaining(task.name));
|
||||
const confirmationMessage = confirm.mock.calls[0]?.[0] || '';
|
||||
expect(confirmationMessage).toContain('Source: MySQL 生产库 / sales');
|
||||
expect(confirmationMessage).toContain('Target: PostgreSQL 数仓 / warehouse / ods');
|
||||
expect(resolveCapability).toHaveBeenCalledWith(expect.objectContaining({
|
||||
id: task.id,
|
||||
revision: task.revision,
|
||||
}));
|
||||
expect(startTask).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ id: task.id, revision: task.revision }),
|
||||
expect.objectContaining({ taskId: task.id }),
|
||||
);
|
||||
expect(
|
||||
renderer.root.findByProps({ 'data-data-sync-run-history': 'true' }),
|
||||
).toBeTruthy();
|
||||
});
|
||||
|
||||
it('loads each scheduled task history when the global run history is capped', async () => {
|
||||
const task = {
|
||||
...buildTask(),
|
||||
lifecycle: 'enabled' as const,
|
||||
trigger: {
|
||||
mode: 'interval' as const,
|
||||
intervalSeconds: 300,
|
||||
timezone: 'Asia/Shanghai',
|
||||
},
|
||||
};
|
||||
const buriedRun: DataSyncRunRecord = {
|
||||
id: 'scheduled-run-beyond-global-cap',
|
||||
taskId: task.id,
|
||||
taskName: task.name,
|
||||
status: 'failed',
|
||||
trigger: 'schedule',
|
||||
attempt: 1,
|
||||
resumable: true,
|
||||
message: 'target write failed after global history cap',
|
||||
startedAt: '2026-08-08T01:00:00.000Z',
|
||||
finishedAt: '2026-08-08T01:01:00.000Z',
|
||||
rowsRead: 10,
|
||||
rowsWritten: 9,
|
||||
rowsFailed: 1,
|
||||
throughput: 9,
|
||||
checkpoint: 'orders:9',
|
||||
};
|
||||
const noisyRuns: DataSyncRunRecord[] = Array.from({ length: 200 }, (_, index) => ({
|
||||
...buriedRun,
|
||||
id: `newer-unrelated-run-${index}`,
|
||||
taskId: `unrelated-task-${index}`,
|
||||
taskName: `Unrelated task ${index}`,
|
||||
status: 'succeeded',
|
||||
trigger: 'manual',
|
||||
resumable: false,
|
||||
message: '',
|
||||
startedAt: '2026-08-08T02:00:00.000Z',
|
||||
finishedAt: '2026-08-08T02:01:00.000Z',
|
||||
rowsRead: 1,
|
||||
rowsWritten: 1,
|
||||
rowsFailed: 0,
|
||||
throughput: 1,
|
||||
checkpoint: '',
|
||||
}));
|
||||
const baseGateway = createStaticDataSyncWorkbenchGateway({
|
||||
tasks: [task],
|
||||
runs: [...noisyRuns, buriedRun],
|
||||
});
|
||||
const listRuns = vi.fn(async (taskId?: string) =>
|
||||
taskId
|
||||
? baseGateway.listRuns(taskId)
|
||||
: (await baseGateway.listRuns()).filter((run) => run.id !== buriedRun.id).slice(0, 200),
|
||||
);
|
||||
const retryRun = vi.fn(baseGateway.retryRun.bind(baseGateway));
|
||||
const gateway = { ...baseGateway, listRuns, retryRun };
|
||||
const renderer = TestRenderer.create(
|
||||
<DataSyncWorkbenchShell initialTasks={[task]} gateway={gateway} locale="en-US" />,
|
||||
);
|
||||
|
||||
await act(async () => {
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
act(() => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Schedules'))!
|
||||
.props.onClick();
|
||||
});
|
||||
|
||||
const row = renderer.root.findByProps({ 'data-task-id': task.id });
|
||||
expect(listRuns).toHaveBeenCalledWith(task.id);
|
||||
expect(
|
||||
row.findAllByProps({ children: buriedRun.message }),
|
||||
).toHaveLength(1);
|
||||
|
||||
await act(async () => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('View run'))!
|
||||
.props.onClick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(
|
||||
renderer.root
|
||||
.findByProps({ 'data-selected': 'true' })
|
||||
.findAllByType('td')[0]!
|
||||
.children,
|
||||
).toContain(buriedRun.id);
|
||||
|
||||
act(() => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Runs'))!
|
||||
.props.onClick();
|
||||
});
|
||||
await act(async () => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Refresh'))!
|
||||
.props.onClick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(
|
||||
renderer.root
|
||||
.findByProps({ 'data-selected': 'true' })
|
||||
.findAllByType('td')[0]!
|
||||
.children,
|
||||
).toContain(buriedRun.id);
|
||||
|
||||
await act(async () => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Retry run'))!
|
||||
.props.onClick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
expect(retryRun).toHaveBeenCalledWith(buriedRun.id);
|
||||
});
|
||||
|
||||
const scheduleRunPreflightCases: Array<{
|
||||
name: string;
|
||||
fixtures: Pick<
|
||||
StaticDataSyncGatewayFixtures,
|
||||
'extraPreflightIssues' | 'approvalRequiredByTask'
|
||||
>;
|
||||
}> = [
|
||||
{
|
||||
name: 'blocked preflight',
|
||||
fixtures: {
|
||||
extraPreflightIssues: [{
|
||||
id: 'target-unavailable',
|
||||
code: 'route_unsupported',
|
||||
severity: 'blocker' as const,
|
||||
stage: 'endpoints' as const,
|
||||
}],
|
||||
},
|
||||
},
|
||||
{
|
||||
name: 'required approval',
|
||||
fixtures: {
|
||||
approvalRequiredByTask: { 'orders-task': true },
|
||||
},
|
||||
},
|
||||
];
|
||||
|
||||
it.each(scheduleRunPreflightCases)('does not start a schedule run with $name', async ({ fixtures }) => {
|
||||
const task = {
|
||||
...buildTask(),
|
||||
lifecycle: 'enabled' as const,
|
||||
trigger: {
|
||||
mode: 'interval' as const,
|
||||
intervalSeconds: 300,
|
||||
timezone: 'Asia/Shanghai',
|
||||
},
|
||||
};
|
||||
const baseGateway = createStaticDataSyncWorkbenchGateway({
|
||||
tasks: [task],
|
||||
capabilities: {
|
||||
[task.id]: {
|
||||
level: 'full',
|
||||
canExecute: true,
|
||||
supportsAutoCreate: true,
|
||||
supportsCdc: false,
|
||||
},
|
||||
},
|
||||
...fixtures,
|
||||
});
|
||||
const startTask = vi.fn(baseGateway.startTask.bind(baseGateway));
|
||||
const gateway = { ...baseGateway, startTask };
|
||||
vi.stubGlobal('confirm', vi.fn(() => true));
|
||||
const renderer = TestRenderer.create(
|
||||
<DataSyncWorkbenchShell initialTasks={[task]} gateway={gateway} locale="en-US" />,
|
||||
);
|
||||
|
||||
await act(async () => {
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
act(() => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Schedules'))!
|
||||
.props.onClick();
|
||||
});
|
||||
await act(async () => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Run now'))!
|
||||
.props.onClick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(startTask).not.toHaveBeenCalled();
|
||||
expect(
|
||||
renderer.root.findByProps({ 'data-data-sync-preflight': 'true' }),
|
||||
).toBeTruthy();
|
||||
});
|
||||
|
||||
it('routes a paused schedule into preflight instead of bypassing the runnable lifecycle gate', async () => {
|
||||
const task = {
|
||||
...buildTask(),
|
||||
lifecycle: 'paused' as const,
|
||||
trigger: {
|
||||
mode: 'interval' as const,
|
||||
intervalSeconds: 300,
|
||||
timezone: 'Asia/Shanghai',
|
||||
},
|
||||
};
|
||||
const baseGateway = createStaticDataSyncWorkbenchGateway({
|
||||
tasks: [task],
|
||||
capabilities: {
|
||||
[task.id]: {
|
||||
level: 'full',
|
||||
canExecute: true,
|
||||
supportsAutoCreate: true,
|
||||
supportsCdc: false,
|
||||
},
|
||||
},
|
||||
});
|
||||
const preflightTask = vi.fn(baseGateway.preflightTask.bind(baseGateway));
|
||||
const startTask = vi.fn(baseGateway.startTask.bind(baseGateway));
|
||||
const gateway = { ...baseGateway, preflightTask, startTask };
|
||||
vi.stubGlobal('confirm', vi.fn(() => true));
|
||||
const renderer = TestRenderer.create(
|
||||
<DataSyncWorkbenchShell initialTasks={[task]} gateway={gateway} locale="en-US" />,
|
||||
);
|
||||
|
||||
await act(async () => {
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
act(() => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Schedules'))!
|
||||
.props.onClick();
|
||||
});
|
||||
await act(async () => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Run now'))!
|
||||
.props.onClick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(preflightTask).toHaveBeenCalledWith(expect.objectContaining({
|
||||
id: task.id,
|
||||
lifecycle: 'paused',
|
||||
}));
|
||||
expect(startTask).not.toHaveBeenCalled();
|
||||
expect(
|
||||
renderer.root.findByProps({ 'data-data-sync-preflight': 'true' }),
|
||||
).toBeTruthy();
|
||||
});
|
||||
|
||||
it('opens a failed schedule run in history where the existing retry flow remains available', async () => {
|
||||
const task = {
|
||||
...buildTask(),
|
||||
lifecycle: 'enabled' as const,
|
||||
trigger: {
|
||||
mode: 'interval' as const,
|
||||
intervalSeconds: 300,
|
||||
timezone: 'Asia/Shanghai',
|
||||
},
|
||||
};
|
||||
const failedRun: DataSyncRunRecord = {
|
||||
id: 'failed-schedule-run',
|
||||
taskId: task.id,
|
||||
taskName: task.name,
|
||||
status: 'failed',
|
||||
trigger: 'schedule',
|
||||
attempt: 1,
|
||||
resumable: true,
|
||||
message: 'target write failed',
|
||||
startedAt: '2026-08-08T01:00:00.000Z',
|
||||
finishedAt: '2026-08-08T01:01:00.000Z',
|
||||
rowsRead: 10,
|
||||
rowsWritten: 9,
|
||||
rowsFailed: 1,
|
||||
throughput: 9,
|
||||
checkpoint: 'orders:9',
|
||||
};
|
||||
const baseGateway = createStaticDataSyncWorkbenchGateway({
|
||||
tasks: [task],
|
||||
runs: [failedRun],
|
||||
});
|
||||
const retryRun = vi.fn(baseGateway.retryRun.bind(baseGateway));
|
||||
const listSchedules = vi.fn(baseGateway.listSchedules.bind(baseGateway));
|
||||
const gateway = { ...baseGateway, listSchedules, retryRun };
|
||||
const renderer = TestRenderer.create(
|
||||
<DataSyncWorkbenchShell initialTasks={[task]} gateway={gateway} locale="en-US" />,
|
||||
);
|
||||
|
||||
await act(async () => {
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
act(() => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Schedules'))!
|
||||
.props.onClick();
|
||||
});
|
||||
await act(async () => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('View run'))!
|
||||
.props.onClick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(
|
||||
renderer.root.findByProps({ 'data-data-sync-run-history': 'true' }),
|
||||
).toBeTruthy();
|
||||
expect(
|
||||
renderer.root
|
||||
.findByProps({ 'data-selected': 'true' })
|
||||
.findAllByType('td')[0]!
|
||||
.children,
|
||||
).toContain(failedRun.id);
|
||||
await act(async () => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Retry run'))!
|
||||
.props.onClick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
expect(retryRun).toHaveBeenCalledWith(failedRun.id);
|
||||
expect(listSchedules.mock.calls.length).toBeGreaterThan(1);
|
||||
act(() => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Schedules'))!
|
||||
.props.onClick();
|
||||
});
|
||||
expect(
|
||||
renderer.root
|
||||
.findByProps({ 'data-task-id': task.id })
|
||||
.findByProps({ 'data-state': 'queued' }),
|
||||
).toBeTruthy();
|
||||
});
|
||||
|
||||
it('invalidates an approval in the UI when a lifecycle-only change keeps the same definition hash', async () => {
|
||||
const task = reviseDataSyncTask(buildTask(), {
|
||||
lifecycle: 'ready',
|
||||
trigger: {
|
||||
mode: 'interval',
|
||||
intervalSeconds: 300,
|
||||
timezone: 'Asia/Shanghai',
|
||||
},
|
||||
});
|
||||
const baseGateway = createStaticDataSyncWorkbenchGateway({
|
||||
tasks: [task],
|
||||
capabilities: {
|
||||
[task.id]: {
|
||||
level: 'full',
|
||||
canExecute: true,
|
||||
supportsAutoCreate: true,
|
||||
supportsCdc: false,
|
||||
},
|
||||
},
|
||||
approvalRequiredByTask: { [task.id]: true },
|
||||
// The backend approval scope also covers lifecycle, while this fixture
|
||||
// holds the hash constant to expose the UI's former hash-only reuse.
|
||||
definitionHashByTask: { [task.id]: 'same-definition-hash' },
|
||||
});
|
||||
const beginApproval = vi.fn(async (_task, preflight) => ({
|
||||
taskId: preflight.taskId,
|
||||
definitionHash: preflight.definitionHash,
|
||||
taskRevision: preflight.taskRevision,
|
||||
notBefore: '2020-01-01T00:00:00.000Z',
|
||||
expiresAt: '2030-08-08T00:02:00.000Z',
|
||||
}));
|
||||
const approveTask = vi.fn(async (_task, preflight) => ({
|
||||
taskId: preflight.taskId,
|
||||
definitionHash: preflight.definitionHash,
|
||||
taskRevision: preflight.taskRevision,
|
||||
expiresAt: '2030-08-08T00:10:00.000Z',
|
||||
}));
|
||||
const gateway = { ...baseGateway, beginApproval, approveTask };
|
||||
const renderer = TestRenderer.create(
|
||||
<DataSyncWorkbenchShell initialTasks={[task]} gateway={gateway} locale="en-US" />,
|
||||
);
|
||||
|
||||
await act(async () => {
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
await act(async () => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Run preflight'))!
|
||||
.props.onClick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
await act(async () => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Begin server 10-second confirmation'))!
|
||||
.props.onClick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
await act(async () => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Confirm production write and grant token'))!
|
||||
.props.onClick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
expect(
|
||||
renderer.root
|
||||
.findAllByType('strong')
|
||||
.some((node) => node.children.includes('One-time production authorization granted')),
|
||||
).toBe(true);
|
||||
|
||||
act(() => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Enable schedule'))!
|
||||
.props.onClick();
|
||||
});
|
||||
await act(async () => {
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Run preflight'))!
|
||||
.props.onClick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(
|
||||
renderer.root
|
||||
.findAllByType('strong')
|
||||
.some((node) => node.children.includes('One-time production authorization granted')),
|
||||
).toBe(false);
|
||||
expect(
|
||||
renderer.root
|
||||
.findAllByType('button')
|
||||
.find((button) => button.children.includes('Save draft'))!.props.disabled,
|
||||
).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import React, { useEffect, useMemo, useRef, useState } from 'react';
|
||||
import React, { useCallback, useEffect, useMemo, useRef, useState } from 'react';
|
||||
|
||||
import {
|
||||
DataSyncCdcView,
|
||||
@@ -17,6 +17,7 @@ import {
|
||||
type DataSyncWorkbenchGateway,
|
||||
} from './gateway';
|
||||
import {
|
||||
aggregateDataSyncScheduleSummaries,
|
||||
canStartDataSyncTask,
|
||||
createDataSyncTaskDraft,
|
||||
isDataSyncPreflightCurrent,
|
||||
@@ -46,6 +47,14 @@ import './DataSyncWorkbench.css';
|
||||
|
||||
type WorkbenchView = 'tasks' | 'runs' | 'schedules' | 'cdc';
|
||||
|
||||
type DataSyncOperationalState = {
|
||||
tasks: DataSyncTaskDefinition[];
|
||||
runs: DataSyncRunRecord[];
|
||||
scheduleRuns: DataSyncRunRecord[];
|
||||
schedules: DataSyncScheduleSummary[];
|
||||
cdcSources: DataSyncCdcSourceStatus[];
|
||||
};
|
||||
|
||||
const EMPTY_CAPABILITY: DataSyncRouteCapability = {
|
||||
level: 'unknown',
|
||||
canExecute: false,
|
||||
@@ -184,6 +193,7 @@ export const DataSyncWorkbenchShell: React.FC<DataSyncWorkbenchShellProps> = ({
|
||||
const [errorRows, setErrorRows] = useState<DataSyncErrorRow[]>([]);
|
||||
const [checkpoint, setCheckpoint] = useState<DataSyncCheckpointSummary | null>(null);
|
||||
const runStatusesRef = useRef<Map<string, DataSyncRunRecord['status']>>(new Map());
|
||||
const scheduleActionBusyRef = useRef(false);
|
||||
|
||||
const selectedTask = tasks.find((task) => task.id === selectedTaskId) || null;
|
||||
const checkpointTask = checkpoint
|
||||
@@ -212,31 +222,84 @@ export const DataSyncWorkbenchShell: React.FC<DataSyncWorkbenchShellProps> = ({
|
||||
);
|
||||
}, [search, tasks]);
|
||||
|
||||
const scheduleImpactScope = (task: DataSyncTaskDefinition): string => {
|
||||
const endpoint = (
|
||||
value: DataSyncTaskDefinition['source'],
|
||||
fallback: string,
|
||||
) => {
|
||||
const connection = value.connectionName || value.connectionId || fallback;
|
||||
const location = [value.database, value.schema].filter(Boolean).join(' / ');
|
||||
return location ? `${connection} / ${location}` : connection;
|
||||
};
|
||||
return [
|
||||
`${t('route.source')}: ${endpoint(task.source, t('route.pending_source'))}`,
|
||||
`${t('route.target')}: ${endpoint(task.target, t('route.pending_target'))}`,
|
||||
].join('\n');
|
||||
};
|
||||
|
||||
const loadOperationalState = useCallback(
|
||||
async (): Promise<DataSyncOperationalState> => {
|
||||
// Wails derives schedules and CDC sources from its cached task projection,
|
||||
// so refresh tasks first and use that same snapshot for every view.
|
||||
const loadedTasks = await gatewayRef.current!.listTasks();
|
||||
const scheduledTasks = loadedTasks.filter(
|
||||
(task) => task.trigger.mode !== 'manual',
|
||||
);
|
||||
const [loadedRuns, loadedSchedules, loadedSources, loadedScheduleRuns] = await Promise.all([
|
||||
gatewayRef.current!.listRuns(),
|
||||
gatewayRef.current!.listSchedules(),
|
||||
gatewayRef.current!.listCdcSources(),
|
||||
Promise.all(
|
||||
scheduledTasks.map((task) => gatewayRef.current!.listRuns(task.id)),
|
||||
),
|
||||
]);
|
||||
return {
|
||||
tasks: loadedTasks,
|
||||
runs: loadedRuns,
|
||||
scheduleRuns: loadedScheduleRuns.flat(),
|
||||
schedules: loadedSchedules,
|
||||
cdcSources: loadedSources,
|
||||
};
|
||||
},
|
||||
[],
|
||||
);
|
||||
|
||||
const applyOperationalState = useCallback(
|
||||
(snapshot: DataSyncOperationalState, includeTasks = false) => {
|
||||
if (includeTasks && snapshot.tasks.length > 0) {
|
||||
setTasks(snapshot.tasks);
|
||||
setSelectedTaskId((current) =>
|
||||
snapshot.tasks.some((task) => task.id === current)
|
||||
? current
|
||||
: snapshot.tasks[0]?.id || '',
|
||||
);
|
||||
}
|
||||
setRuns(snapshot.runs);
|
||||
setSchedules(
|
||||
aggregateDataSyncScheduleSummaries(
|
||||
snapshot.tasks,
|
||||
snapshot.scheduleRuns,
|
||||
snapshot.schedules,
|
||||
),
|
||||
);
|
||||
setCdcSources(snapshot.cdcSources);
|
||||
},
|
||||
[],
|
||||
);
|
||||
|
||||
const refreshScheduleState = useCallback(async () => {
|
||||
const snapshot = await loadOperationalState();
|
||||
applyOperationalState(snapshot);
|
||||
return snapshot.tasks;
|
||||
}, [applyOperationalState, loadOperationalState]);
|
||||
|
||||
useEffect(() => {
|
||||
let active = true;
|
||||
void gatewayRef.current!
|
||||
.listTasks()
|
||||
.then(async (loadedTasks) => {
|
||||
const [loadedRuns, loadedSchedules, loadedSources] = await Promise.all([
|
||||
gatewayRef.current!.listRuns(),
|
||||
gatewayRef.current!.listSchedules(),
|
||||
gatewayRef.current!.listCdcSources(),
|
||||
]);
|
||||
return [loadedTasks, loadedRuns, loadedSchedules, loadedSources] as const;
|
||||
})
|
||||
.then(([loadedTasks, loadedRuns, loadedSchedules, loadedSources]) => {
|
||||
if (!active) return;
|
||||
if (loadedTasks.length > 0) {
|
||||
setTasks(loadedTasks);
|
||||
setSelectedTaskId((current) =>
|
||||
loadedTasks.some((task) => task.id === current)
|
||||
? current
|
||||
: loadedTasks[0].id,
|
||||
);
|
||||
void loadOperationalState()
|
||||
.then((snapshot) => {
|
||||
if (active) {
|
||||
applyOperationalState(snapshot, true);
|
||||
}
|
||||
setRuns(loadedRuns);
|
||||
setSchedules(loadedSchedules);
|
||||
setCdcSources(loadedSources);
|
||||
})
|
||||
.catch((error) => {
|
||||
if (active) {
|
||||
@@ -246,7 +309,7 @@ export const DataSyncWorkbenchShell: React.FC<DataSyncWorkbenchShellProps> = ({
|
||||
return () => {
|
||||
active = false;
|
||||
};
|
||||
}, []);
|
||||
}, [applyOperationalState, loadOperationalState]);
|
||||
|
||||
useEffect(() => {
|
||||
if (!selectedTask) {
|
||||
@@ -395,6 +458,9 @@ export const DataSyncWorkbenchShell: React.FC<DataSyncWorkbenchShellProps> = ({
|
||||
delete next[saved.id];
|
||||
return next;
|
||||
});
|
||||
if (saved.trigger.mode !== 'manual') {
|
||||
await refreshScheduleState();
|
||||
}
|
||||
} catch (error) {
|
||||
setOperationError(error instanceof Error ? error.message : String(error));
|
||||
} finally {
|
||||
@@ -411,14 +477,22 @@ export const DataSyncWorkbenchShell: React.FC<DataSyncWorkbenchShellProps> = ({
|
||||
setPreflights((current) => ({ ...current, [selectedTask.id]: snapshot }));
|
||||
setApprovals((current) => {
|
||||
const approval = current[selectedTask.id];
|
||||
if (!approval || approval.definitionHash === snapshot.definitionHash) return current;
|
||||
if (
|
||||
!approval ||
|
||||
(approval.definitionHash === snapshot.definitionHash &&
|
||||
approval.taskRevision === snapshot.taskRevision)
|
||||
) return current;
|
||||
const next = { ...current };
|
||||
delete next[selectedTask.id];
|
||||
return next;
|
||||
});
|
||||
setApprovalChallenges((current) => {
|
||||
const challenge = current[selectedTask.id];
|
||||
if (!challenge || challenge.definitionHash === snapshot.definitionHash) {
|
||||
if (
|
||||
!challenge ||
|
||||
(challenge.definitionHash === snapshot.definitionHash &&
|
||||
challenge.taskRevision === snapshot.taskRevision)
|
||||
) {
|
||||
return current;
|
||||
}
|
||||
const next = { ...current };
|
||||
@@ -464,11 +538,18 @@ export const DataSyncWorkbenchShell: React.FC<DataSyncWorkbenchShellProps> = ({
|
||||
}
|
||||
};
|
||||
|
||||
const selectRun = async (runId: string) => {
|
||||
const selectRun = async (runId: string, taskIdHint = '') => {
|
||||
setSelectedRunId(runId);
|
||||
const run = runs.find((item) => item.id === runId);
|
||||
setOperationError('');
|
||||
try {
|
||||
let run = runs.find((item) => item.id === runId);
|
||||
if (!run && taskIdHint) {
|
||||
const taskRuns = await gatewayRef.current!.listRuns(taskIdHint);
|
||||
run = taskRuns.find((item) => item.id === runId);
|
||||
if (run) {
|
||||
setRuns((current) => [run!, ...current.filter((item) => item.id !== runId)]);
|
||||
}
|
||||
}
|
||||
const [rows, loadedCheckpoint] = await Promise.all([
|
||||
gatewayRef.current!.listErrorRows(runId),
|
||||
run ? gatewayRef.current!.getCheckpoint(run.taskId) : Promise.resolve(null),
|
||||
@@ -528,8 +609,30 @@ export const DataSyncWorkbenchShell: React.FC<DataSyncWorkbenchShellProps> = ({
|
||||
setOperationBusy('refresh-runs');
|
||||
setOperationError('');
|
||||
try {
|
||||
setRuns(await gatewayRef.current!.listRuns());
|
||||
if (selectedRunId) await selectRun(selectedRunId);
|
||||
// The global endpoint is capped. Keep the selected schedule run visible
|
||||
// by using its task-scoped history when the capped result omits it.
|
||||
let taskIdHint = selectedRunId
|
||||
? runs.find((run) => run.id === selectedRunId)?.taskId ||
|
||||
schedules.find((schedule) => schedule.latestRun?.id === selectedRunId)
|
||||
?.taskId ||
|
||||
''
|
||||
: '';
|
||||
const refreshedRuns = await gatewayRef.current!.listRuns();
|
||||
const selectedFromGlobal = refreshedRuns.find((run) => run.id === selectedRunId);
|
||||
if (selectedFromGlobal) taskIdHint = selectedFromGlobal.taskId;
|
||||
let nextRuns = refreshedRuns;
|
||||
if (selectedRunId && !selectedFromGlobal && taskIdHint) {
|
||||
const taskRuns = await gatewayRef.current!.listRuns(taskIdHint);
|
||||
const selectedFromTask = taskRuns.find((run) => run.id === selectedRunId);
|
||||
if (selectedFromTask) {
|
||||
nextRuns = [
|
||||
selectedFromTask,
|
||||
...refreshedRuns.filter((run) => run.id !== selectedFromTask.id),
|
||||
];
|
||||
}
|
||||
}
|
||||
setRuns(nextRuns);
|
||||
if (selectedRunId) await selectRun(selectedRunId, taskIdHint);
|
||||
} catch (error) {
|
||||
setOperationError(error instanceof Error ? error.message : String(error));
|
||||
} finally {
|
||||
@@ -558,6 +661,7 @@ export const DataSyncWorkbenchShell: React.FC<DataSyncWorkbenchShellProps> = ({
|
||||
: await gatewayRef.current!.retryRun(runId);
|
||||
setRuns((current) => [run, ...current.filter((item) => item.id !== run.id)]);
|
||||
}
|
||||
await refreshScheduleState();
|
||||
} catch (error) {
|
||||
setOperationError(error instanceof Error ? error.message : String(error));
|
||||
} finally {
|
||||
@@ -652,8 +756,9 @@ export const DataSyncWorkbenchShell: React.FC<DataSyncWorkbenchShellProps> = ({
|
||||
|
||||
const refreshSchedules = async () => {
|
||||
setOperationBusy('refresh-schedules');
|
||||
setOperationError('');
|
||||
try {
|
||||
setSchedules(await gatewayRef.current!.listSchedules());
|
||||
await refreshScheduleState();
|
||||
} catch (error) {
|
||||
setOperationError(error instanceof Error ? error.message : String(error));
|
||||
} finally {
|
||||
@@ -661,6 +766,189 @@ export const DataSyncWorkbenchShell: React.FC<DataSyncWorkbenchShellProps> = ({
|
||||
}
|
||||
};
|
||||
|
||||
const replaceTaskState = (saved: DataSyncTaskDefinition) => {
|
||||
setTasks((current) =>
|
||||
current.map((task) => (task.id === saved.id ? saved : task)),
|
||||
);
|
||||
setDirtyTaskIds((current) => {
|
||||
const next = new Set(current);
|
||||
next.delete(saved.id);
|
||||
return next;
|
||||
});
|
||||
setPreflights((current) => {
|
||||
if (!current[saved.id]) return current;
|
||||
const next = { ...current };
|
||||
delete next[saved.id];
|
||||
return next;
|
||||
});
|
||||
setApprovals((current) => {
|
||||
if (!current[saved.id]) return current;
|
||||
const next = { ...current };
|
||||
delete next[saved.id];
|
||||
return next;
|
||||
});
|
||||
setApprovalChallenges((current) => {
|
||||
if (!current[saved.id]) return current;
|
||||
const next = { ...current };
|
||||
delete next[saved.id];
|
||||
return next;
|
||||
});
|
||||
};
|
||||
|
||||
const showScheduleTaskPreflight = (
|
||||
task: DataSyncTaskDefinition,
|
||||
preflight: DataSyncPreflightSnapshot,
|
||||
resolvedCapability: DataSyncRouteCapability,
|
||||
) => {
|
||||
setTasks((current) =>
|
||||
current.map((item) => (item.id === task.id ? task : item)),
|
||||
);
|
||||
setSelectedTaskId(task.id);
|
||||
setPreflights((current) => ({ ...current, [task.id]: preflight }));
|
||||
setApprovals((current) => {
|
||||
const approval = current[task.id];
|
||||
if (
|
||||
!approval ||
|
||||
(approval.definitionHash === preflight.definitionHash &&
|
||||
approval.taskRevision === preflight.taskRevision)
|
||||
) {
|
||||
return current;
|
||||
}
|
||||
const next = { ...current };
|
||||
delete next[task.id];
|
||||
return next;
|
||||
});
|
||||
setApprovalChallenges((current) => {
|
||||
const challenge = current[task.id];
|
||||
if (
|
||||
!challenge ||
|
||||
(challenge.definitionHash === preflight.definitionHash &&
|
||||
challenge.taskRevision === preflight.taskRevision)
|
||||
) {
|
||||
return current;
|
||||
}
|
||||
const next = { ...current };
|
||||
delete next[task.id];
|
||||
return next;
|
||||
});
|
||||
setCapability(resolvedCapability);
|
||||
setActiveView('tasks');
|
||||
setActiveStage('preflight');
|
||||
};
|
||||
|
||||
const toggleSchedule = async (schedule: DataSyncScheduleSummary) => {
|
||||
if (scheduleActionBusyRef.current) return;
|
||||
scheduleActionBusyRef.current = true;
|
||||
const enabling = !schedule.enabled;
|
||||
const action = enabling ? 'enable' : 'disable';
|
||||
const confirmKey = enabling
|
||||
? 'schedules.confirm_enable'
|
||||
: 'schedules.confirm_disable';
|
||||
setOperationBusy(`${action}:${schedule.taskId}`);
|
||||
setOperationError('');
|
||||
try {
|
||||
const latestTasks = await gatewayRef.current!.listTasks();
|
||||
const current = latestTasks.find((task) => task.id === schedule.taskId);
|
||||
if (!current) throw new Error(t('schedules.task_missing'));
|
||||
if (dirtyTaskIds.has(current.id)) {
|
||||
throw new Error(t('schedules.unsaved_edits', { task: current.name }));
|
||||
}
|
||||
if (!globalThis.confirm(t(confirmKey, {
|
||||
task: current.name,
|
||||
scope: scheduleImpactScope(current),
|
||||
}))) {
|
||||
return;
|
||||
}
|
||||
// Store.PutJob accepts the persisted revision and advances it itself.
|
||||
const next: DataSyncTaskDefinition = {
|
||||
...current,
|
||||
lifecycle: enabling ? 'enabled' : 'paused',
|
||||
};
|
||||
if (enabling) {
|
||||
const [preflight, resolvedCapability] = await Promise.all([
|
||||
gatewayRef.current!.preflightTask(next),
|
||||
gatewayRef.current!.resolveCapability(next),
|
||||
]);
|
||||
if (
|
||||
preflight.status === 'blocked' ||
|
||||
preflight.approvalRequired ||
|
||||
!resolvedCapability.canExecute
|
||||
) {
|
||||
showScheduleTaskPreflight(next, preflight, resolvedCapability);
|
||||
setTasks((state) =>
|
||||
state.map((task) => (task.id === next.id ? next : task)),
|
||||
);
|
||||
setDirtyTaskIds((state) => new Set(state).add(next.id));
|
||||
setOperationError(
|
||||
t('schedules.preflight_required', { task: next.name }),
|
||||
);
|
||||
return;
|
||||
}
|
||||
}
|
||||
const saved = await gatewayRef.current!.saveTask(next);
|
||||
replaceTaskState(saved);
|
||||
await refreshScheduleState();
|
||||
} catch (error) {
|
||||
setOperationError(error instanceof Error ? error.message : String(error));
|
||||
} finally {
|
||||
scheduleActionBusyRef.current = false;
|
||||
setOperationBusy('');
|
||||
}
|
||||
};
|
||||
|
||||
const runScheduleNow = async (schedule: DataSyncScheduleSummary) => {
|
||||
if (scheduleActionBusyRef.current) return;
|
||||
scheduleActionBusyRef.current = true;
|
||||
setOperationBusy(`run-now:${schedule.taskId}`);
|
||||
setOperationError('');
|
||||
try {
|
||||
const latestTasks = await gatewayRef.current!.listTasks();
|
||||
const task = latestTasks.find((item) => item.id === schedule.taskId);
|
||||
if (!task) throw new Error(t('schedules.task_missing'));
|
||||
if (dirtyTaskIds.has(task.id)) {
|
||||
throw new Error(t('schedules.unsaved_edits', { task: task.name }));
|
||||
}
|
||||
if (!globalThis.confirm(t('schedules.confirm_run_now', {
|
||||
task: task.name,
|
||||
scope: scheduleImpactScope(task),
|
||||
}))) {
|
||||
return;
|
||||
}
|
||||
const [preflight, resolvedCapability] = await Promise.all([
|
||||
gatewayRef.current!.preflightTask(task),
|
||||
gatewayRef.current!.resolveCapability(task),
|
||||
]);
|
||||
showScheduleTaskPreflight(task, preflight, resolvedCapability);
|
||||
if (
|
||||
!resolvedCapability.canExecute ||
|
||||
!canStartDataSyncTask(task, preflight, approvals[task.id] || null)
|
||||
) {
|
||||
setOperationError(t('schedules.preflight_required', { task: task.name }));
|
||||
return;
|
||||
}
|
||||
const run = await gatewayRef.current!.startTask(
|
||||
task,
|
||||
preflight,
|
||||
);
|
||||
setRuns((current) => [run, ...current.filter((item) => item.id !== run.id)]);
|
||||
await refreshScheduleState();
|
||||
setSelectedRunId(run.id);
|
||||
setActiveView('runs');
|
||||
} catch (error) {
|
||||
setOperationError(error instanceof Error ? error.message : String(error));
|
||||
} finally {
|
||||
scheduleActionBusyRef.current = false;
|
||||
setOperationBusy('');
|
||||
}
|
||||
};
|
||||
|
||||
const viewScheduleRun = (runId: string) => {
|
||||
const taskId = schedules.find((schedule) => schedule.latestRun?.id === runId)?.taskId || '';
|
||||
setSelectedRunId(runId);
|
||||
setActiveView('runs');
|
||||
void selectRun(runId, taskId);
|
||||
};
|
||||
|
||||
const refreshCdc = async () => {
|
||||
setOperationBusy('refresh-cdc');
|
||||
try {
|
||||
@@ -704,6 +992,7 @@ export const DataSyncWorkbenchShell: React.FC<DataSyncWorkbenchShellProps> = ({
|
||||
Boolean(
|
||||
selectedApproval &&
|
||||
selectedApproval.definitionHash === selectedPreflight.definitionHash &&
|
||||
selectedApproval.taskRevision === selectedPreflight.taskRevision &&
|
||||
Date.parse(selectedApproval.expiresAt) > Date.now(),
|
||||
)))),
|
||||
);
|
||||
@@ -944,6 +1233,10 @@ export const DataSyncWorkbenchShell: React.FC<DataSyncWorkbenchShellProps> = ({
|
||||
t={t}
|
||||
refreshing={operationBusy === 'refresh-schedules'}
|
||||
onRefresh={() => void refreshSchedules()}
|
||||
busyAction={operationBusy}
|
||||
onToggle={(schedule) => void toggleSchedule(schedule)}
|
||||
onRunNow={(schedule) => void runScheduleNow(schedule)}
|
||||
onViewRun={viewScheduleRun}
|
||||
/>
|
||||
) : null}
|
||||
{activeView === 'cdc' ? (
|
||||
|
||||
@@ -5,6 +5,7 @@ import {
|
||||
createDataSyncTableMapping,
|
||||
createDataSyncTaskDraft,
|
||||
reviseDataSyncTask,
|
||||
type DataSyncRunRecord,
|
||||
} from './model';
|
||||
|
||||
const configuredTask = () => {
|
||||
@@ -56,9 +57,269 @@ describe('static data sync workbench gateway', () => {
|
||||
});
|
||||
expect(await gateway.listRuns(task.id)).toHaveLength(1);
|
||||
|
||||
const renamed = reviseDataSyncTask(task, { name: 'Renamed migration' });
|
||||
await gateway.saveTask(renamed);
|
||||
expect(await gateway.listTasks()).toContainEqual(renamed);
|
||||
const renamed = { ...task, name: 'Renamed migration' };
|
||||
const saved = await gateway.saveTask(renamed);
|
||||
expect(saved).toMatchObject({
|
||||
id: task.id,
|
||||
name: renamed.name,
|
||||
revision: task.revision + 1,
|
||||
});
|
||||
expect(await gateway.listTasks()).toContainEqual(saved);
|
||||
});
|
||||
|
||||
it('keeps paused schedule state and active runs in sync with the persisted task revision', async () => {
|
||||
const task = {
|
||||
...configuredTask(),
|
||||
lifecycle: 'enabled' as const,
|
||||
trigger: {
|
||||
mode: 'interval' as const,
|
||||
intervalSeconds: 300,
|
||||
timezone: 'Asia/Shanghai',
|
||||
},
|
||||
};
|
||||
const queued: DataSyncRunRecord = {
|
||||
id: 'queued-run',
|
||||
taskId: task.id,
|
||||
taskName: task.name,
|
||||
status: 'queued',
|
||||
trigger: 'schedule',
|
||||
attempt: 1,
|
||||
resumable: false,
|
||||
message: 'waiting to start',
|
||||
startedAt: '2026-08-08T00:00:00.000Z',
|
||||
finishedAt: '',
|
||||
rowsRead: 0,
|
||||
rowsWritten: 0,
|
||||
rowsFailed: 0,
|
||||
throughput: 0,
|
||||
checkpoint: '',
|
||||
};
|
||||
const running: DataSyncRunRecord = {
|
||||
...queued,
|
||||
id: 'running-run',
|
||||
status: 'running',
|
||||
message: '',
|
||||
startedAt: '2026-08-08T00:10:00.000Z',
|
||||
};
|
||||
const cancelling: DataSyncRunRecord = {
|
||||
...queued,
|
||||
id: 'cancelling-run',
|
||||
status: 'cancelling',
|
||||
message: 'operator requested cancellation',
|
||||
startedAt: '2026-08-08T00:20:00.000Z',
|
||||
};
|
||||
const gateway = createStaticDataSyncWorkbenchGateway({
|
||||
tasks: [task],
|
||||
runs: [queued, running, cancelling],
|
||||
schedules: [
|
||||
{
|
||||
id: `${task.id}:schedule`,
|
||||
taskId: task.id,
|
||||
taskName: task.name,
|
||||
enabled: true,
|
||||
expression: '300s',
|
||||
timezone: 'Asia/Shanghai',
|
||||
nextRunAt: '2026-08-08T02:00:00.000Z',
|
||||
},
|
||||
],
|
||||
now: () => '2026-08-08T01:00:00.000Z',
|
||||
});
|
||||
|
||||
const paused = await gateway.saveTask({ ...task, lifecycle: 'paused' });
|
||||
|
||||
expect(paused).toMatchObject({
|
||||
lifecycle: 'paused',
|
||||
revision: task.revision + 1,
|
||||
updatedAt: '2026-08-08T01:00:00.000Z',
|
||||
});
|
||||
expect(await gateway.listRuns(task.id)).toEqual([
|
||||
expect.objectContaining({
|
||||
id: queued.id,
|
||||
status: 'canceled',
|
||||
finishedAt: '2026-08-08T01:00:00.000Z',
|
||||
message: 'canceled because task was paused',
|
||||
}),
|
||||
expect.objectContaining({
|
||||
id: running.id,
|
||||
status: 'cancelling',
|
||||
finishedAt: '',
|
||||
message: 'cancellation requested because task was paused',
|
||||
}),
|
||||
expect.objectContaining({
|
||||
id: cancelling.id,
|
||||
status: 'cancelling',
|
||||
finishedAt: '',
|
||||
message: 'operator requested cancellation',
|
||||
}),
|
||||
]);
|
||||
expect(await gateway.listSchedules()).toEqual([
|
||||
expect.objectContaining({
|
||||
taskId: task.id,
|
||||
revision: paused.revision,
|
||||
lifecycle: 'paused',
|
||||
enabled: false,
|
||||
nextRunAt: '',
|
||||
}),
|
||||
]);
|
||||
|
||||
const enabled = await gateway.saveTask({ ...paused, lifecycle: 'enabled' });
|
||||
expect(await gateway.listSchedules()).toEqual([
|
||||
expect.objectContaining({
|
||||
taskId: task.id,
|
||||
revision: enabled.revision,
|
||||
lifecycle: 'enabled',
|
||||
enabled: true,
|
||||
nextRunAt: '',
|
||||
}),
|
||||
]);
|
||||
await expect(gateway.saveTask({ ...task, name: 'Stale schedule' })).rejects.toThrow(
|
||||
'revision changed',
|
||||
);
|
||||
expect(await gateway.listTasks()).toEqual([enabled]);
|
||||
});
|
||||
|
||||
it('rejects an immediate run that uses the task revision before a schedule change', async () => {
|
||||
const task = {
|
||||
...configuredTask(),
|
||||
lifecycle: 'enabled' as const,
|
||||
trigger: {
|
||||
mode: 'interval' as const,
|
||||
intervalSeconds: 300,
|
||||
timezone: 'Asia/Shanghai',
|
||||
},
|
||||
};
|
||||
const gateway = createStaticDataSyncWorkbenchGateway({
|
||||
tasks: [task],
|
||||
capabilities: {
|
||||
[task.id]: {
|
||||
level: 'full',
|
||||
canExecute: true,
|
||||
supportsAutoCreate: true,
|
||||
supportsCdc: false,
|
||||
},
|
||||
},
|
||||
});
|
||||
const preflight = await gateway.preflightTask(task);
|
||||
|
||||
await gateway.saveTask({ ...task, lifecycle: 'paused' });
|
||||
|
||||
await expect(gateway.startTask(task, preflight)).rejects.toThrow('revision changed');
|
||||
expect(await gateway.listRuns(task.id)).toEqual([]);
|
||||
});
|
||||
|
||||
it('records a schedule-list immediate run as a manual trigger', async () => {
|
||||
const task = {
|
||||
...configuredTask(),
|
||||
lifecycle: 'enabled' as const,
|
||||
trigger: {
|
||||
mode: 'cron' as const,
|
||||
expression: '0 * * * *',
|
||||
timezone: 'UTC',
|
||||
overlap: 'skip' as const,
|
||||
},
|
||||
};
|
||||
const gateway = createStaticDataSyncWorkbenchGateway({
|
||||
tasks: [task],
|
||||
capabilities: {
|
||||
[task.id]: {
|
||||
level: 'full',
|
||||
canExecute: true,
|
||||
supportsAutoCreate: true,
|
||||
supportsCdc: false,
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const run = await gateway.startTask(task, await gateway.preflightTask(task));
|
||||
|
||||
expect(run).toMatchObject({ taskId: task.id, trigger: 'manual' });
|
||||
});
|
||||
|
||||
it('archives scheduled tasks by canceling active runs and removing their list projections', async () => {
|
||||
const task = {
|
||||
...configuredTask(),
|
||||
lifecycle: 'enabled' as const,
|
||||
trigger: {
|
||||
mode: 'cron' as const,
|
||||
expression: '0 * * * *',
|
||||
timezone: 'UTC',
|
||||
overlap: 'skip' as const,
|
||||
},
|
||||
};
|
||||
const gateway = createStaticDataSyncWorkbenchGateway({
|
||||
tasks: [task],
|
||||
runs: [
|
||||
{
|
||||
id: 'archive-queued-run',
|
||||
taskId: task.id,
|
||||
taskName: task.name,
|
||||
status: 'queued',
|
||||
trigger: 'schedule',
|
||||
attempt: 1,
|
||||
resumable: false,
|
||||
message: '',
|
||||
startedAt: '',
|
||||
finishedAt: '',
|
||||
rowsRead: 0,
|
||||
rowsWritten: 0,
|
||||
rowsFailed: 0,
|
||||
throughput: 0,
|
||||
checkpoint: '',
|
||||
},
|
||||
{
|
||||
id: 'archive-streaming-run',
|
||||
taskId: task.id,
|
||||
taskName: task.name,
|
||||
status: 'streaming',
|
||||
trigger: 'continuous',
|
||||
attempt: 1,
|
||||
resumable: false,
|
||||
message: '',
|
||||
startedAt: '2026-08-08T00:30:00.000Z',
|
||||
finishedAt: '',
|
||||
rowsRead: 0,
|
||||
rowsWritten: 0,
|
||||
rowsFailed: 0,
|
||||
throughput: 0,
|
||||
checkpoint: '',
|
||||
},
|
||||
],
|
||||
schedules: [
|
||||
{
|
||||
id: `${task.id}:schedule`,
|
||||
taskId: task.id,
|
||||
taskName: task.name,
|
||||
enabled: true,
|
||||
expression: '0 * * * *',
|
||||
timezone: 'UTC',
|
||||
nextRunAt: '2026-08-08T02:00:00.000Z',
|
||||
},
|
||||
],
|
||||
now: () => '2026-08-08T01:00:00.000Z',
|
||||
});
|
||||
|
||||
const archived = await gateway.saveTask({ ...task, lifecycle: 'archived' });
|
||||
|
||||
expect(archived).toMatchObject({
|
||||
lifecycle: 'archived',
|
||||
revision: task.revision + 1,
|
||||
});
|
||||
expect(await gateway.listTasks()).toEqual([]);
|
||||
expect(await gateway.listSchedules()).toEqual([]);
|
||||
expect(await gateway.listRuns(task.id)).toEqual([
|
||||
expect.objectContaining({
|
||||
id: 'archive-queued-run',
|
||||
status: 'canceled',
|
||||
finishedAt: '2026-08-08T01:00:00.000Z',
|
||||
message: 'canceled because task was archived',
|
||||
}),
|
||||
expect.objectContaining({
|
||||
id: 'archive-streaming-run',
|
||||
status: 'cancelling',
|
||||
finishedAt: '',
|
||||
message: 'cancellation requested because task was archived',
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it('fails closed with an explicit warning when no backend capability is injected', async () => {
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import {
|
||||
aggregateDataSyncScheduleSummaries,
|
||||
resolveDataSyncPreflightStatus,
|
||||
validateDataSyncTask,
|
||||
type DataSyncApprovalChallenge,
|
||||
@@ -174,6 +175,20 @@ const copy = <T,>(value: T): T => {
|
||||
return JSON.parse(JSON.stringify(value)) as T;
|
||||
};
|
||||
|
||||
const INACTIVE_RUN_STATUSES = new Set<DataSyncRunRecord['status']>([
|
||||
'queued',
|
||||
'paused',
|
||||
]);
|
||||
|
||||
const CANCELLING_RUN_STATUSES = new Set<DataSyncRunRecord['status']>([
|
||||
'running',
|
||||
'cancelling',
|
||||
'preflighting',
|
||||
'snapshotting',
|
||||
'catching_up',
|
||||
'streaming',
|
||||
]);
|
||||
|
||||
const unresolvedCapability: DataSyncRouteCapability = {
|
||||
level: 'unknown',
|
||||
canExecute: false,
|
||||
@@ -206,6 +221,46 @@ export const createStaticDataSyncWorkbenchGateway = (
|
||||
const fields = copy(fixtures.fieldsByObject || DEFAULT_FIELDS);
|
||||
const now = fixtures.now || (() => new Date().toISOString());
|
||||
|
||||
const visibleTasks = (): DataSyncTaskDefinition[] =>
|
||||
Array.from(taskMap.values()).filter((task) => task.lifecycle !== 'archived');
|
||||
|
||||
const synchronizeScheduleFixture = (task: DataSyncTaskDefinition) => {
|
||||
const index = schedules.findIndex((schedule) => schedule.taskId === task.id);
|
||||
if (task.lifecycle === 'archived' || task.trigger.mode === 'manual') {
|
||||
if (index >= 0) schedules.splice(index, 1);
|
||||
return;
|
||||
}
|
||||
if (index < 0) return;
|
||||
const schedule = schedules[index];
|
||||
schedules[index] = {
|
||||
...schedule,
|
||||
taskName: task.name,
|
||||
revision: task.revision,
|
||||
lifecycle: task.lifecycle,
|
||||
enabled: task.lifecycle === 'enabled',
|
||||
nextRunAt: task.lifecycle === 'enabled' ? schedule.nextRunAt : '',
|
||||
};
|
||||
};
|
||||
|
||||
const cancelInactiveTaskRuns = (
|
||||
task: DataSyncTaskDefinition,
|
||||
canceledAt: string,
|
||||
) => {
|
||||
const canceledMessage = `canceled because task was ${task.lifecycle}`;
|
||||
const cancellingMessage = `cancellation requested because task was ${task.lifecycle}`;
|
||||
for (const run of runs) {
|
||||
if (run.taskId !== task.id) continue;
|
||||
if (INACTIVE_RUN_STATUSES.has(run.status)) {
|
||||
run.status = 'canceled';
|
||||
run.finishedAt = canceledAt;
|
||||
run.message = canceledMessage;
|
||||
} else if (CANCELLING_RUN_STATUSES.has(run.status)) {
|
||||
run.status = 'cancelling';
|
||||
if (!run.message) run.message = cancellingMessage;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
return {
|
||||
capabilities: { errorRowRetry: false },
|
||||
async listSavedConnections() {
|
||||
@@ -231,11 +286,25 @@ export const createStaticDataSyncWorkbenchGateway = (
|
||||
return (fields[exactKey] || fields[withoutSchemaKey] || []).map(copy);
|
||||
},
|
||||
async listTasks() {
|
||||
return Array.from(taskMap.values()).map(copy);
|
||||
return visibleTasks().map(copy);
|
||||
},
|
||||
async saveTask(task) {
|
||||
const saved = copy(task);
|
||||
const current = taskMap.get(task.id);
|
||||
if (current && current.revision !== task.revision) {
|
||||
throw new Error('data sync task revision changed');
|
||||
}
|
||||
const savedAt = now();
|
||||
const saved = {
|
||||
...copy(task),
|
||||
revision: current ? current.revision + 1 : 1,
|
||||
createdAt: current?.createdAt || task.createdAt || savedAt,
|
||||
updatedAt: savedAt,
|
||||
};
|
||||
taskMap.set(saved.id, saved);
|
||||
synchronizeScheduleFixture(saved);
|
||||
if (saved.lifecycle === 'paused' || saved.lifecycle === 'archived') {
|
||||
cancelInactiveTaskRuns(saved, savedAt);
|
||||
}
|
||||
return copy(saved);
|
||||
},
|
||||
async resolveCapability(task) {
|
||||
@@ -282,10 +351,15 @@ export const createStaticDataSyncWorkbenchGateway = (
|
||||
.map(copy);
|
||||
},
|
||||
async startTask(task, preflight) {
|
||||
const current = taskMap.get(task.id);
|
||||
if (!current) throw new Error('data sync task not found');
|
||||
if (current.revision !== task.revision) {
|
||||
throw new Error('data sync task revision changed');
|
||||
}
|
||||
if (
|
||||
(task.lifecycle !== 'ready' && task.lifecycle !== 'enabled') ||
|
||||
(current.lifecycle !== 'ready' && current.lifecycle !== 'enabled') ||
|
||||
preflight.taskId !== task.id ||
|
||||
preflight.taskRevision !== task.revision ||
|
||||
preflight.taskRevision !== current.revision ||
|
||||
preflight.status === 'blocked' ||
|
||||
preflight.approvalRequired !== false
|
||||
) {
|
||||
@@ -293,19 +367,15 @@ export const createStaticDataSyncWorkbenchGateway = (
|
||||
}
|
||||
const startedAt = now();
|
||||
const run: DataSyncRunRecord = {
|
||||
id: `${task.id}:run:${startedAt}`,
|
||||
taskId: task.id,
|
||||
taskName: task.name,
|
||||
status: task.kind === 'cdc' ? 'streaming' : 'queued',
|
||||
trigger:
|
||||
task.trigger.mode === 'manual'
|
||||
? 'manual'
|
||||
: task.trigger.mode === 'continuous'
|
||||
? 'continuous'
|
||||
: 'schedule',
|
||||
id: `${current.id}:run:${startedAt}`,
|
||||
taskId: current.id,
|
||||
taskName: current.name,
|
||||
status: current.kind === 'cdc' ? 'streaming' : 'queued',
|
||||
trigger: 'manual',
|
||||
attempt: 1,
|
||||
resumable: false,
|
||||
message: '',
|
||||
queuedAt: startedAt,
|
||||
startedAt,
|
||||
finishedAt: '',
|
||||
rowsRead: 0,
|
||||
@@ -321,7 +391,7 @@ export const createStaticDataSyncWorkbenchGateway = (
|
||||
return (errors[runId] || []).map(copy);
|
||||
},
|
||||
async listSchedules() {
|
||||
return schedules.map(copy);
|
||||
return aggregateDataSyncScheduleSummaries(visibleTasks(), runs, schedules).map(copy);
|
||||
},
|
||||
async listCdcAdapters() {
|
||||
return cdcAdapters.slice();
|
||||
@@ -364,6 +434,7 @@ export const createStaticDataSyncWorkbenchGateway = (
|
||||
status: 'queued' as const,
|
||||
trigger: 'resume' as const,
|
||||
attempt: previous.attempt + 1,
|
||||
queuedAt: now(),
|
||||
startedAt: '',
|
||||
finishedAt: '',
|
||||
};
|
||||
@@ -379,6 +450,7 @@ export const createStaticDataSyncWorkbenchGateway = (
|
||||
status: 'queued' as const,
|
||||
trigger: 'retry' as const,
|
||||
attempt: previous.attempt + 1,
|
||||
queuedAt: now(),
|
||||
startedAt: '',
|
||||
finishedAt: '',
|
||||
};
|
||||
|
||||
@@ -2,6 +2,7 @@ import { describe, expect, it } from 'vitest';
|
||||
|
||||
import {
|
||||
autoMatchDataSyncFields,
|
||||
aggregateDataSyncScheduleSummaries,
|
||||
canUseDataSyncRowErrorIsolation,
|
||||
canStartDataSyncTask,
|
||||
buildDataSyncMappingsFromSelection,
|
||||
@@ -10,11 +11,355 @@ import {
|
||||
isDataSyncPreflightCurrent,
|
||||
resolveDataSyncPreflightStatus,
|
||||
reviseDataSyncTask,
|
||||
summarizeDataSyncRunMessage,
|
||||
validateDataSyncTask,
|
||||
type DataSyncPreflightSnapshot,
|
||||
} from './model';
|
||||
|
||||
describe('data sync task model', () => {
|
||||
it('redacts and bounds schedule error summaries', () => {
|
||||
const message =
|
||||
'connect failed: postgres://alice:secret@example.test/db?token=abc password=top-secret; ' +
|
||||
'x'.repeat(300);
|
||||
|
||||
const summary = summarizeDataSyncRunMessage(message, 80);
|
||||
|
||||
expect(summary).not.toContain('secret');
|
||||
expect(summary).not.toContain('abc');
|
||||
expect(summary).not.toContain('top-secret');
|
||||
expect(summary).toContain('[REDACTED]');
|
||||
expect(summary.length).toBeLessThanOrEqual(80);
|
||||
expect(summary.endsWith('…')).toBe(true);
|
||||
});
|
||||
|
||||
it('redacts quoted credential values that contain spaces', () => {
|
||||
const summary = summarizeDataSyncRunMessage(
|
||||
'target rejected password="top secret"; token: \'one time token\'; api_key = "blue green"',
|
||||
);
|
||||
|
||||
expect(summary).toBe(
|
||||
'target rejected password=[REDACTED]; token: [REDACTED]; api_key = [REDACTED]',
|
||||
);
|
||||
});
|
||||
|
||||
it('redacts unquoted credential values with spaces up to the next field', () => {
|
||||
const summary = summarizeDataSyncRunMessage(
|
||||
'target rejected password=top secret; retryable=true; token=one time token',
|
||||
);
|
||||
|
||||
expect(summary).toBe(
|
||||
'target rejected password=[REDACTED]; retryable=true; token=[REDACTED]',
|
||||
);
|
||||
expect(summary).not.toContain('top secret');
|
||||
expect(summary).not.toContain('one time token');
|
||||
});
|
||||
|
||||
it('redacts authorization schemes and URL userinfo edge cases', () => {
|
||||
const summary = summarizeDataSyncRunMessage(
|
||||
'Authorization: Basic dXNlcjpzZWNyZXQ=; ' +
|
||||
'postgres://:empty-user-password@example.test/db; ' +
|
||||
'postgres://alice:p@ss@example.test/db',
|
||||
);
|
||||
|
||||
expect(summary).not.toContain('dXNlcjpzZWNyZXQ=');
|
||||
expect(summary).not.toContain('empty-user-password');
|
||||
expect(summary).not.toContain('p@ss');
|
||||
expect(summary).toContain('Authorization: [REDACTED]');
|
||||
expect(summary).toContain('postgres://[REDACTED]@example.test/db');
|
||||
});
|
||||
|
||||
it('redacts quoted query credentials without leaving value fragments', () => {
|
||||
const summary = summarizeDataSyncRunMessage(
|
||||
'request failed: https://example.test/api?password="top secret"&token=one%20time',
|
||||
);
|
||||
|
||||
expect(summary).toBe(
|
||||
'request failed: https://example.test/api?password=[REDACTED]&token=[REDACTED]',
|
||||
);
|
||||
expect(summary).not.toContain('top secret');
|
||||
expect(summary).not.toContain('one%20time');
|
||||
});
|
||||
|
||||
it('redacts composite OAuth and client credential keys in schedule summaries', () => {
|
||||
const summary = summarizeDataSyncRunMessage(
|
||||
'authorization failed: access_token=access-value; refresh_token: refresh-value; client_secret = client-value; Access-Token: header-value',
|
||||
);
|
||||
|
||||
for (const value of [
|
||||
'access-value',
|
||||
'refresh-value',
|
||||
'client-value',
|
||||
'header-value',
|
||||
]) {
|
||||
expect(summary).not.toContain(value);
|
||||
}
|
||||
expect((summary.match(/\[REDACTED\]/g) || [])).toHaveLength(4);
|
||||
});
|
||||
|
||||
it('redacts credential values in JSON-shaped error messages', () => {
|
||||
const summary = summarizeDataSyncRunMessage(
|
||||
'{"access_token":"access-value","refresh_token":"refresh-value","client_secret":"client-value","password":"top-secret"}',
|
||||
);
|
||||
|
||||
for (const value of [
|
||||
'access-value',
|
||||
'refresh-value',
|
||||
'client-value',
|
||||
'top-secret',
|
||||
]) {
|
||||
expect(summary).not.toContain(value);
|
||||
}
|
||||
expect((summary.match(/\[REDACTED\]/g) || [])).toHaveLength(4);
|
||||
});
|
||||
|
||||
it('aggregates the latest run and lifecycle state for scheduled tasks', () => {
|
||||
const scheduled = reviseDataSyncTask(
|
||||
createDataSyncTaskDraft({
|
||||
id: 'scheduled-task',
|
||||
kind: 'reconcile',
|
||||
name: 'Scheduled orders',
|
||||
}),
|
||||
{
|
||||
lifecycle: 'paused',
|
||||
trigger: {
|
||||
mode: 'cron',
|
||||
expression: '0 * * * *',
|
||||
timezone: 'Asia/Shanghai',
|
||||
overlap: 'skip',
|
||||
},
|
||||
},
|
||||
);
|
||||
const manual = createDataSyncTaskDraft({
|
||||
id: 'manual-task',
|
||||
kind: 'migration',
|
||||
name: 'Manual migration',
|
||||
});
|
||||
const summaries = aggregateDataSyncScheduleSummaries(
|
||||
[scheduled, manual],
|
||||
[
|
||||
{
|
||||
id: 'old-failure',
|
||||
taskId: scheduled.id,
|
||||
taskName: scheduled.name,
|
||||
status: 'failed',
|
||||
trigger: 'schedule',
|
||||
attempt: 1,
|
||||
resumable: false,
|
||||
message: 'old failure',
|
||||
startedAt: '2026-08-08T00:00:00.000Z',
|
||||
finishedAt: '2026-08-08T00:01:00.000Z',
|
||||
rowsRead: 0,
|
||||
rowsWritten: 0,
|
||||
rowsFailed: 1,
|
||||
throughput: 0,
|
||||
checkpoint: '',
|
||||
},
|
||||
{
|
||||
id: 'latest-failure',
|
||||
taskId: scheduled.id,
|
||||
taskName: scheduled.name,
|
||||
status: 'failed',
|
||||
trigger: 'schedule',
|
||||
attempt: 1,
|
||||
resumable: true,
|
||||
message: 'password=latest-secret',
|
||||
startedAt: '2026-08-08T01:00:00.000Z',
|
||||
finishedAt: '2026-08-08T01:02:00.000Z',
|
||||
rowsRead: 3,
|
||||
rowsWritten: 2,
|
||||
rowsFailed: 1,
|
||||
throughput: 1,
|
||||
checkpoint: '',
|
||||
},
|
||||
],
|
||||
[
|
||||
{
|
||||
id: 'scheduled-task:schedule',
|
||||
taskId: scheduled.id,
|
||||
taskName: scheduled.name,
|
||||
enabled: true,
|
||||
expression: 'old-expression',
|
||||
timezone: 'UTC',
|
||||
nextRunAt: '2026-08-08T02:00:00.000Z',
|
||||
},
|
||||
],
|
||||
);
|
||||
|
||||
expect(summaries).toHaveLength(1);
|
||||
expect(summaries[0]).toMatchObject({
|
||||
taskId: scheduled.id,
|
||||
lifecycle: 'paused',
|
||||
enabled: false,
|
||||
expression: '0 * * * *',
|
||||
timezone: 'Asia/Shanghai',
|
||||
nextRunAt: '',
|
||||
latestRun: {
|
||||
id: 'latest-failure',
|
||||
status: 'failed',
|
||||
errorSummary: 'password=[REDACTED]',
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it('shows sanitized messages for canceled and interrupted latest runs', () => {
|
||||
const scheduled = reviseDataSyncTask(
|
||||
createDataSyncTaskDraft({
|
||||
id: 'scheduled-terminal-task',
|
||||
kind: 'reconcile',
|
||||
name: 'Scheduled terminal state',
|
||||
}),
|
||||
{
|
||||
lifecycle: 'enabled',
|
||||
trigger: {
|
||||
mode: 'interval',
|
||||
intervalSeconds: 60,
|
||||
timezone: 'UTC',
|
||||
},
|
||||
},
|
||||
);
|
||||
|
||||
const [summary] = aggregateDataSyncScheduleSummaries([scheduled], [
|
||||
{
|
||||
id: 'interrupted-run',
|
||||
taskId: scheduled.id,
|
||||
taskName: scheduled.name,
|
||||
status: 'interrupted',
|
||||
trigger: 'schedule',
|
||||
attempt: 1,
|
||||
resumable: true,
|
||||
message: 'authorization=secret-value',
|
||||
startedAt: '2026-08-08T02:00:00.000Z',
|
||||
finishedAt: '2026-08-08T02:01:00.000Z',
|
||||
rowsRead: 1,
|
||||
rowsWritten: 0,
|
||||
rowsFailed: 1,
|
||||
throughput: 0,
|
||||
checkpoint: '',
|
||||
},
|
||||
]);
|
||||
|
||||
expect(summary.latestRun?.errorSummary).toBe('authorization=[REDACTED]');
|
||||
});
|
||||
|
||||
it('uses the latest run start time when completion order differs', () => {
|
||||
const scheduled = reviseDataSyncTask(
|
||||
createDataSyncTaskDraft({
|
||||
id: 'scheduled-order-task',
|
||||
kind: 'reconcile',
|
||||
name: 'Scheduled order check',
|
||||
}),
|
||||
{
|
||||
lifecycle: 'enabled',
|
||||
trigger: {
|
||||
mode: 'interval',
|
||||
intervalSeconds: 300,
|
||||
timezone: 'UTC',
|
||||
},
|
||||
},
|
||||
);
|
||||
const summaries = aggregateDataSyncScheduleSummaries([scheduled], [
|
||||
{
|
||||
id: 'older-start-finished-last',
|
||||
taskId: scheduled.id,
|
||||
taskName: scheduled.name,
|
||||
status: 'failed',
|
||||
trigger: 'schedule',
|
||||
attempt: 1,
|
||||
resumable: false,
|
||||
message: 'older start',
|
||||
startedAt: '2026-08-08T01:00:00.000Z',
|
||||
finishedAt: '2026-08-08T03:00:00.000Z',
|
||||
rowsRead: 0,
|
||||
rowsWritten: 0,
|
||||
rowsFailed: 1,
|
||||
throughput: 0,
|
||||
checkpoint: '',
|
||||
},
|
||||
{
|
||||
id: 'newer-start-finished-first',
|
||||
taskId: scheduled.id,
|
||||
taskName: scheduled.name,
|
||||
status: 'succeeded',
|
||||
trigger: 'schedule',
|
||||
attempt: 1,
|
||||
resumable: false,
|
||||
message: '',
|
||||
startedAt: '2026-08-08T02:00:00.000Z',
|
||||
finishedAt: '2026-08-08T02:30:00.000Z',
|
||||
rowsRead: 1,
|
||||
rowsWritten: 1,
|
||||
rowsFailed: 0,
|
||||
throughput: 1,
|
||||
checkpoint: '',
|
||||
},
|
||||
]);
|
||||
|
||||
expect(summaries[0]?.latestRun).toMatchObject({
|
||||
id: 'newer-start-finished-first',
|
||||
status: 'succeeded',
|
||||
});
|
||||
});
|
||||
|
||||
it('treats a queued retry as the latest schedule result', () => {
|
||||
const scheduled = reviseDataSyncTask(
|
||||
createDataSyncTaskDraft({
|
||||
id: 'scheduled-retry-task',
|
||||
kind: 'reconcile',
|
||||
name: 'Scheduled retry check',
|
||||
}),
|
||||
{
|
||||
lifecycle: 'enabled',
|
||||
trigger: {
|
||||
mode: 'interval',
|
||||
intervalSeconds: 300,
|
||||
timezone: 'UTC',
|
||||
},
|
||||
},
|
||||
);
|
||||
const summaries = aggregateDataSyncScheduleSummaries([scheduled], [
|
||||
{
|
||||
id: 'failed-before-retry',
|
||||
taskId: scheduled.id,
|
||||
taskName: scheduled.name,
|
||||
status: 'failed',
|
||||
trigger: 'schedule',
|
||||
attempt: 1,
|
||||
resumable: true,
|
||||
message: 'target write failed',
|
||||
startedAt: '2026-08-08T01:00:00.000Z',
|
||||
finishedAt: '2026-08-08T01:01:00.000Z',
|
||||
rowsRead: 1,
|
||||
rowsWritten: 0,
|
||||
rowsFailed: 1,
|
||||
throughput: 0,
|
||||
checkpoint: '',
|
||||
},
|
||||
{
|
||||
id: 'queued-retry',
|
||||
taskId: scheduled.id,
|
||||
taskName: scheduled.name,
|
||||
status: 'queued',
|
||||
trigger: 'retry',
|
||||
attempt: 2,
|
||||
resumable: false,
|
||||
message: '',
|
||||
startedAt: '',
|
||||
finishedAt: '',
|
||||
queuedAt: '2026-08-08T02:00:00.000Z',
|
||||
rowsRead: 0,
|
||||
rowsWritten: 0,
|
||||
rowsFailed: 0,
|
||||
throughput: 0,
|
||||
checkpoint: '',
|
||||
},
|
||||
]);
|
||||
|
||||
expect(summaries[0]?.latestRun).toMatchObject({
|
||||
id: 'queued-retry',
|
||||
status: 'queued',
|
||||
});
|
||||
});
|
||||
|
||||
it('creates versioned defaults for compare and CDC tasks', () => {
|
||||
const compare = createDataSyncTaskDraft({
|
||||
id: 'compare-1',
|
||||
|
||||
@@ -279,13 +279,21 @@ export type DataSyncPreflightSnapshot = {
|
||||
};
|
||||
|
||||
export type DataSyncApprovalChallenge = {
|
||||
/** ID of the exact task covered by the server countdown. */
|
||||
taskId: string;
|
||||
definitionHash: string;
|
||||
/** Revision of the exact task definition approved for the countdown. */
|
||||
taskRevision: number;
|
||||
notBefore: string;
|
||||
expiresAt: string;
|
||||
};
|
||||
|
||||
export type DataSyncApprovalGrant = {
|
||||
/** ID of the exact task covered by the one-time token. */
|
||||
taskId: string;
|
||||
definitionHash: string;
|
||||
/** Revision of the exact task definition covered by the one-time token. */
|
||||
taskRevision: number;
|
||||
expiresAt: string;
|
||||
};
|
||||
|
||||
@@ -324,6 +332,8 @@ export type DataSyncRunRecord = {
|
||||
attempt: number;
|
||||
resumable: boolean;
|
||||
message: string;
|
||||
/** Queue time is retained for runs that have not started yet. */
|
||||
queuedAt?: string;
|
||||
startedAt: string;
|
||||
finishedAt: string;
|
||||
rowsRead: number;
|
||||
@@ -350,10 +360,137 @@ export type DataSyncScheduleSummary = {
|
||||
id: string;
|
||||
taskId: string;
|
||||
taskName: string;
|
||||
revision?: number;
|
||||
lifecycle?: DataSyncTaskLifecycle;
|
||||
enabled: boolean;
|
||||
expression: string;
|
||||
timezone: string;
|
||||
nextRunAt: string;
|
||||
latestRun?: DataSyncScheduleRunSummary | null;
|
||||
};
|
||||
|
||||
/** The bounded, credential-free run projection shown in the schedule list. */
|
||||
export type DataSyncScheduleRunSummary = {
|
||||
id: string;
|
||||
status: DataSyncRunStatus;
|
||||
startedAt: string;
|
||||
finishedAt: string;
|
||||
errorSummary: string;
|
||||
};
|
||||
|
||||
const DATA_SYNC_SCHEDULE_ERROR_MAX_LENGTH = 240;
|
||||
|
||||
/**
|
||||
* Redacts common credential-shaped values before a run message can reach the
|
||||
* schedule overview. The backend remains authoritative; this is a second
|
||||
* presentation-boundary guard for static and older gateway responses.
|
||||
*/
|
||||
export const summarizeDataSyncRunMessage = (
|
||||
message: string,
|
||||
maxLength = DATA_SYNC_SCHEDULE_ERROR_MAX_LENGTH,
|
||||
): string => {
|
||||
const limit = Number.isFinite(maxLength) && maxLength > 0
|
||||
? Math.floor(maxLength)
|
||||
: DATA_SYNC_SCHEDULE_ERROR_MAX_LENGTH;
|
||||
let summary = String(message || '')
|
||||
.replace(/\b(?:Basic|Bearer)\s+[A-Za-z0-9._~+/=-]+/gi, '[REDACTED]')
|
||||
.replace(
|
||||
/([a-z][a-z0-9+.-]*:\/\/)[^\s/?#]*@/gi,
|
||||
'$1[REDACTED]@',
|
||||
)
|
||||
.replace(
|
||||
/([?&](?:password|passwd|pwd|(?:access|refresh|id)[_-]?token|token|(?:client|app)[_-]?secret|secret|api[_-]?key|authorization|dsn|credential)(?:=|%3d))(?:"(?:\\.|[^"\\])*"|'(?:\\.|[^'\\])*'|[^&#\s;]+)/gi,
|
||||
'$1[REDACTED]',
|
||||
)
|
||||
.replace(
|
||||
/(["'])(password|passwd|pwd|(?:access|refresh|id)[_-]?token|token|(?:client|app)[_-]?secret|secret|api[_-]?key|authorization|dsn|credential)\1(\s*[:=]\s*)(?:"(?:\\.|[^"\\])*"|'(?:\\.|[^'\\])*'|[^,;\s"'}]+)/gi,
|
||||
'$1$2$1$3[REDACTED]',
|
||||
)
|
||||
.replace(
|
||||
/\b(password|passwd|pwd|(?:access|refresh|id)[_-]?token|token|(?:client|app)[_-]?secret|secret|api[_-]?key|authorization|dsn|credential)(\s*[:=]\s*)(?:"[^"]*"|'[^']*'|[^,;&\r\n]+)/gi,
|
||||
'$1$2[REDACTED]',
|
||||
)
|
||||
.replace(/\s+/g, ' ')
|
||||
.trim();
|
||||
if (summary.length <= limit) return summary;
|
||||
return `${summary.slice(0, Math.max(0, limit - 1)).trimEnd()}…`;
|
||||
};
|
||||
|
||||
const scheduleExpression = (trigger: DataSyncTriggerPolicy): string => {
|
||||
switch (trigger.mode) {
|
||||
case 'cron':
|
||||
return trigger.expression;
|
||||
case 'interval':
|
||||
return `${trigger.intervalSeconds}s`;
|
||||
case 'once':
|
||||
return trigger.runAt;
|
||||
case 'continuous':
|
||||
return 'continuous';
|
||||
default:
|
||||
return '';
|
||||
}
|
||||
};
|
||||
|
||||
const scheduleTimezone = (trigger: DataSyncTriggerPolicy): string =>
|
||||
trigger.mode === 'cron' || trigger.mode === 'interval' || trigger.mode === 'once'
|
||||
? trigger.timezone || 'Local'
|
||||
: 'Local';
|
||||
|
||||
const runSortTimestamp = (run: DataSyncRunRecord): number => {
|
||||
const started = Date.parse(run.startedAt);
|
||||
if (Number.isFinite(started)) return started;
|
||||
const queued = Date.parse(run.queuedAt || '');
|
||||
if (Number.isFinite(queued)) return queued;
|
||||
const finished = Date.parse(run.finishedAt);
|
||||
return Number.isFinite(finished) ? finished : 0;
|
||||
};
|
||||
|
||||
/**
|
||||
* Builds schedule rows from the task and run projections when a dedicated
|
||||
* schedule endpoint is unavailable. Existing rows are used to retain the
|
||||
* server-computed next-run timestamp.
|
||||
*/
|
||||
export const aggregateDataSyncScheduleSummaries = (
|
||||
tasks: DataSyncTaskDefinition[],
|
||||
runs: DataSyncRunRecord[] = [],
|
||||
existing: DataSyncScheduleSummary[] = [],
|
||||
): DataSyncScheduleSummary[] => {
|
||||
const existingByTask = new Map(existing.map((item) => [item.taskId, item]));
|
||||
return tasks
|
||||
.filter((task) => task.trigger.mode !== 'manual')
|
||||
.map((task) => {
|
||||
const prior = existingByTask.get(task.id);
|
||||
const latest = runs
|
||||
.filter((run) => run.taskId === task.id)
|
||||
.sort((left, right) => {
|
||||
const timestampDelta = runSortTimestamp(right) - runSortTimestamp(left);
|
||||
return timestampDelta || right.id.localeCompare(left.id);
|
||||
})[0];
|
||||
return {
|
||||
id: prior?.id || `${task.id}:schedule`,
|
||||
taskId: task.id,
|
||||
taskName: task.name,
|
||||
revision: task.revision,
|
||||
lifecycle: task.lifecycle,
|
||||
enabled: task.lifecycle === 'enabled',
|
||||
expression: scheduleExpression(task.trigger),
|
||||
timezone: scheduleTimezone(task.trigger),
|
||||
nextRunAt:
|
||||
task.lifecycle === 'enabled' ? prior?.nextRunAt || '' : '',
|
||||
latestRun: latest
|
||||
? {
|
||||
id: latest.id,
|
||||
status: latest.status,
|
||||
startedAt: latest.startedAt,
|
||||
finishedAt: latest.finishedAt,
|
||||
errorSummary:
|
||||
latest.status !== 'succeeded' && latest.message
|
||||
? summarizeDataSyncRunMessage(latest.message)
|
||||
: '',
|
||||
}
|
||||
: null,
|
||||
};
|
||||
});
|
||||
};
|
||||
|
||||
export type DataSyncCdcSourceStatus = {
|
||||
@@ -878,7 +1015,9 @@ export const canStartDataSyncTask = (
|
||||
preflight?.approvalSatisfied === true ||
|
||||
Boolean(
|
||||
approval &&
|
||||
approval.taskId === task.id &&
|
||||
approval.definitionHash === preflight?.definitionHash &&
|
||||
approval.taskRevision === preflight?.taskRevision &&
|
||||
Date.parse(approval.expiresAt) > now,
|
||||
)) &&
|
||||
(preflight?.status === 'passed' || preflight?.status === 'warning');
|
||||
|
||||
@@ -305,9 +305,28 @@ const zhCN = {
|
||||
'common.dismiss': '关闭',
|
||||
'common.cancel': '取消',
|
||||
'schedules.title': '调度',
|
||||
'schedules.subtitle': '集中查看任务时区、Cron 和下一次执行。',
|
||||
'schedules.subtitle': '集中查看调度状态、最近运行和下一次执行。',
|
||||
'schedules.empty_title': '还没有启用调度',
|
||||
'schedules.empty_desc': '在任务的“触发与增量”阶段选择指定时间或 Cron。',
|
||||
'schedules.task': '任务',
|
||||
'schedules.status': '调度状态',
|
||||
'schedules.trigger': '触发方式',
|
||||
'schedules.next_run': '下次执行',
|
||||
'schedules.latest_run': '最近运行',
|
||||
'schedules.actions': '操作',
|
||||
'schedules.enabled': '已启用',
|
||||
'schedules.disabled': '已停用',
|
||||
'schedules.enable': '启用',
|
||||
'schedules.disable': '停用',
|
||||
'schedules.run_now': '立即运行',
|
||||
'schedules.view_run': '查看运行记录',
|
||||
'schedules.no_runs': '暂无运行记录',
|
||||
'schedules.confirm_disable': '确定停用任务“{task}”吗？\n{scope}\n在途运行会被取消，后续调度不会再触发。',
|
||||
'schedules.confirm_enable': '确定启用任务“{task}”吗？\n{scope}\n后续调度会按当前任务定义执行。',
|
||||
'schedules.confirm_run_now': '确定立即运行任务“{task}”吗？\n{scope}\n这会按当前任务定义写入目标端。',
|
||||
'schedules.preflight_required': '任务“{task}”需要先完成预检与审批，已打开任务检查页面。',
|
||||
'schedules.task_missing': '任务已不存在，请刷新调度清单后重试。',
|
||||
'schedules.unsaved_edits': '任务“{task}”有未保存修改，请先保存或放弃修改后再操作调度。',
|
||||
'cdc.title': 'CDC 源状态',
|
||||
'cdc.subtitle': '状态来自适配器探测和任务 checkpoint；后端未报告延迟时不会推测。',
|
||||
'cdc.empty_title': '还没有可监控的 CDC 源',
|
||||
@@ -727,9 +746,28 @@ const enUS: Record<DataSyncWorkbenchTextKey, string> = {
|
||||
'common.dismiss': 'Dismiss',
|
||||
'common.cancel': 'Cancel',
|
||||
'schedules.title': 'Schedules',
|
||||
'schedules.subtitle': 'Review timezones, Cron expressions, and next runs in one place.',
|
||||
'schedules.subtitle': 'Review schedule state, latest runs, and next runs in one place.',
|
||||
'schedules.empty_title': 'No enabled schedules',
|
||||
'schedules.empty_desc': 'Choose a one-time or Cron trigger in the task editor.',
|
||||
'schedules.task': 'Task',
|
||||
'schedules.status': 'Schedule state',
|
||||
'schedules.trigger': 'Trigger',
|
||||
'schedules.next_run': 'Next run',
|
||||
'schedules.latest_run': 'Latest run',
|
||||
'schedules.actions': 'Actions',
|
||||
'schedules.enabled': 'Enabled',
|
||||
'schedules.disabled': 'Disabled',
|
||||
'schedules.enable': 'Enable',
|
||||
'schedules.disable': 'Disable',
|
||||
'schedules.run_now': 'Run now',
|
||||
'schedules.view_run': 'View run',
|
||||
'schedules.no_runs': 'No run history',
|
||||
'schedules.confirm_disable': 'Disable "{task}"?\n{scope}\nIn-flight runs will be cancelled and future schedules will stop.',
|
||||
'schedules.confirm_enable': 'Enable "{task}"?\n{scope}\nFuture schedules will use the current task definition.',
|
||||
'schedules.confirm_run_now': 'Run "{task}" now?\n{scope}\nThe current task definition may write to the target.',
|
||||
'schedules.preflight_required': 'Task "{task}" needs preflight and approval first. The task checks are now open.',
|
||||
'schedules.task_missing': 'The task no longer exists. Refresh schedules and try again.',
|
||||
'schedules.unsaved_edits': 'Task "{task}" has unsaved edits. Save or discard them before changing its schedule.',
|
||||
'cdc.title': 'CDC source status',
|
||||
'cdc.subtitle': 'Status comes from adapter probes and task checkpoints. Lag is never inferred when the backend does not report it.',
|
||||
'cdc.empty_title': 'No CDC sources to monitor',
|
||||
|
||||
@@ -1081,6 +1081,7 @@ export const decodeRunRecord = (
|
||||
attempt: optionalNumber(run.attempt, 'run.attempt'),
|
||||
resumable: optionalBoolean(run.resumable, 'run.resumable'),
|
||||
message: optionalString(run.message, 'run.message'),
|
||||
queuedAt: fromMillis(run.queuedAt, 'run.queuedAt'),
|
||||
startedAt:
|
||||
fromMillis(run.startedAt, 'run.startedAt') || fromMillis(run.queuedAt, 'run.queuedAt'),
|
||||
finishedAt: fromMillis(run.finishedAt, 'run.finishedAt'),
|
||||
|
||||
@@ -294,7 +294,9 @@ describe('real Wails data sync gateway', () => {
|
||||
|
||||
const challenge = await gateway.beginApproval(task, preflight);
|
||||
expect(challenge).toEqual({
|
||||
taskId: task.id,
|
||||
definitionHash: 'definition-hash',
|
||||
taskRevision: task.revision,
|
||||
notBefore: new Date(NOW + 10_000).toISOString(),
|
||||
expiresAt: new Date(NOW + 120_000).toISOString(),
|
||||
});
|
||||
@@ -307,7 +309,9 @@ describe('real Wails data sync gateway', () => {
|
||||
clock = NOW + 10_000;
|
||||
const grant = await gateway.approveTask(task, preflight);
|
||||
expect(grant).toEqual({
|
||||
taskId: task.id,
|
||||
definitionHash: 'definition-hash',
|
||||
taskRevision: task.revision,
|
||||
expiresAt: new Date(NOW + 600_000).toISOString(),
|
||||
});
|
||||
expect(grant).not.toHaveProperty('token');
|
||||
@@ -334,6 +338,246 @@ describe('real Wails data sync gateway', () => {
|
||||
expect(api.DataSyncRunStart).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('rejects a forged preflight task id before beginning approval', async () => {
|
||||
const task = taskFixture();
|
||||
const api = apiFixture();
|
||||
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => NOW });
|
||||
const preflight = await gateway.preflightTask(task);
|
||||
|
||||
await expect(
|
||||
gateway.beginApproval(task, { ...preflight, taskId: 'other-task' }),
|
||||
).rejects.toThrow('does not match the current task');
|
||||
expect(api.DataSyncJobApprovalBegin).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('rejects a forged preflight revision before beginning approval and clears the challenge', async () => {
|
||||
const task = taskFixture();
|
||||
const api = apiFixture();
|
||||
let clock = NOW;
|
||||
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => clock });
|
||||
const preflight = await gateway.preflightTask(task);
|
||||
await gateway.beginApproval(task, preflight);
|
||||
|
||||
await expect(
|
||||
gateway.beginApproval(task, {
|
||||
...preflight,
|
||||
taskRevision: task.revision + 1,
|
||||
}),
|
||||
).rejects.toThrow('does not match the current task');
|
||||
expect(api.DataSyncJobApprovalBegin).toHaveBeenCalledTimes(1);
|
||||
|
||||
clock = NOW + 10_000;
|
||||
await expect(gateway.approveTask(task, preflight)).rejects.toThrow(
|
||||
'countdown is incomplete',
|
||||
);
|
||||
expect(api.DataSyncJobApprove).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('rejects a stale preflight revision and discards its cached challenge', async () => {
|
||||
const task = taskFixture();
|
||||
const api = apiFixture();
|
||||
let clock = NOW;
|
||||
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => clock });
|
||||
const preflight = await gateway.preflightTask(task);
|
||||
await gateway.beginApproval(task, preflight);
|
||||
|
||||
await expect(
|
||||
gateway.approveTask(task, {
|
||||
...preflight,
|
||||
taskRevision: task.revision + 1,
|
||||
}),
|
||||
).rejects.toThrow('does not match the current task');
|
||||
expect(api.DataSyncJobApprove).not.toHaveBeenCalled();
|
||||
|
||||
clock = NOW + 10_000;
|
||||
await expect(gateway.approveTask(task, preflight)).rejects.toThrow(
|
||||
'countdown is incomplete',
|
||||
);
|
||||
expect(api.DataSyncJobApprove).not.toHaveBeenCalled();
|
||||
|
||||
await gateway.beginApproval(task, preflight);
|
||||
await gateway.approveTask(task, preflight);
|
||||
await expect(
|
||||
gateway.startTask(task, {
|
||||
...preflight,
|
||||
definitionHash: 'forged-definition-hash',
|
||||
}),
|
||||
).rejects.toThrow('preflight is blocked or stale');
|
||||
expect(api.DataSyncRunStart).not.toHaveBeenCalled();
|
||||
await expect(gateway.startTask(task, preflight)).rejects.toThrow(
|
||||
'explicit production approval is required',
|
||||
);
|
||||
});
|
||||
|
||||
it('rejects a forged preflight task id and discards its cached grant', async () => {
|
||||
const task = taskFixture();
|
||||
const api = apiFixture();
|
||||
let clock = NOW;
|
||||
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => clock });
|
||||
const preflight = await gateway.preflightTask(task);
|
||||
await gateway.beginApproval(task, preflight);
|
||||
clock = NOW + 10_000;
|
||||
await gateway.approveTask(task, preflight);
|
||||
|
||||
await expect(
|
||||
gateway.startTask(task, { ...preflight, taskId: 'other-task' }),
|
||||
).rejects.toThrow('does not match the current task');
|
||||
expect(api.DataSyncRunStart).not.toHaveBeenCalled();
|
||||
|
||||
await expect(gateway.startTask(task, preflight)).rejects.toThrow(
|
||||
'explicit production approval is required',
|
||||
);
|
||||
expect(api.DataSyncRunStart).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('rejects a forged preflight task id before approving and clears the challenge', async () => {
|
||||
const task = taskFixture();
|
||||
const api = apiFixture();
|
||||
let clock = NOW;
|
||||
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => clock });
|
||||
const preflight = await gateway.preflightTask(task);
|
||||
await gateway.beginApproval(task, preflight);
|
||||
clock = NOW + 10_000;
|
||||
|
||||
await expect(
|
||||
gateway.approveTask(task, { ...preflight, taskId: 'other-task' }),
|
||||
).rejects.toThrow('does not match the current task');
|
||||
expect(api.DataSyncJobApprove).not.toHaveBeenCalled();
|
||||
|
||||
await expect(gateway.approveTask(task, preflight)).rejects.toThrow(
|
||||
'countdown is incomplete',
|
||||
);
|
||||
expect(api.DataSyncJobApprove).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('rejects a forged preflight revision before starting and clears the grant', async () => {
|
||||
const task = taskFixture();
|
||||
const api = apiFixture();
|
||||
let clock = NOW;
|
||||
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => clock });
|
||||
const preflight = await gateway.preflightTask(task);
|
||||
await gateway.beginApproval(task, preflight);
|
||||
clock = NOW + 10_000;
|
||||
await gateway.approveTask(task, preflight);
|
||||
|
||||
await expect(
|
||||
gateway.startTask(task, {
|
||||
...preflight,
|
||||
taskRevision: task.revision + 1,
|
||||
}),
|
||||
).rejects.toThrow('does not match the current task');
|
||||
expect(api.DataSyncRunStart).not.toHaveBeenCalled();
|
||||
|
||||
await expect(gateway.startTask(task, preflight)).rejects.toThrow(
|
||||
'explicit production approval is required',
|
||||
);
|
||||
expect(api.DataSyncRunStart).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('discards cached authorization when the supplied definition hash changes', async () => {
|
||||
const task = taskFixture();
|
||||
const api = apiFixture();
|
||||
let clock = NOW;
|
||||
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => clock });
|
||||
const preflight = await gateway.preflightTask(task);
|
||||
await gateway.beginApproval(task, preflight);
|
||||
|
||||
await expect(
|
||||
gateway.beginApproval(task, {
|
||||
...preflight,
|
||||
definitionHash: 'forged-definition-hash',
|
||||
}),
|
||||
).rejects.toThrow('approval does not match the current passed preflight');
|
||||
expect(api.DataSyncJobApprovalBegin).toHaveBeenCalledTimes(1);
|
||||
|
||||
clock = NOW + 10_000;
|
||||
await expect(gateway.approveTask(task, preflight)).rejects.toThrow(
|
||||
'countdown is incomplete',
|
||||
);
|
||||
expect(api.DataSyncJobApprove).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('fails closed when a blocked cached preflight is presented as passed', async () => {
|
||||
const task = taskFixture();
|
||||
const api = apiFixture({
|
||||
DataSyncJobPreflight: vi.fn(async (definition) => ({
|
||||
success: false,
|
||||
message: 'preflight blocked',
|
||||
data: {
|
||||
status: 'blocked',
|
||||
definition,
|
||||
definitionHash: 'blocked-definition-hash',
|
||||
approvalRequired: false,
|
||||
capability: {
|
||||
supportLevel: 'full',
|
||||
// Capability resolution alone must not override a blocked preflight.
|
||||
canExecute: true,
|
||||
supportsAutoCreate: true,
|
||||
},
|
||||
issues: [
|
||||
{
|
||||
code: 'target_table_missing',
|
||||
severity: 'blocker',
|
||||
stage: 'delivery',
|
||||
},
|
||||
],
|
||||
checkedAt: NOW,
|
||||
},
|
||||
})),
|
||||
});
|
||||
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => NOW });
|
||||
const blocked = await gateway.preflightTask(task);
|
||||
|
||||
await expect(
|
||||
gateway.startTask(task, { ...blocked, status: 'passed' }),
|
||||
).rejects.toThrow('preflight is blocked or stale');
|
||||
expect(api.DataSyncRunStart).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('keeps an unused production token through an identical fresh preflight', async () => {
|
||||
const task = taskFixture();
|
||||
const api = apiFixture();
|
||||
let clock = NOW;
|
||||
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => clock });
|
||||
const initialPreflight = await gateway.preflightTask(task);
|
||||
|
||||
await gateway.beginApproval(task, initialPreflight);
|
||||
clock = NOW + 10_000;
|
||||
await gateway.approveTask(task, initialPreflight);
|
||||
|
||||
const refreshedPreflight = await gateway.preflightTask(task);
|
||||
|
||||
await expect(gateway.startTask(task, refreshedPreflight)).resolves.toMatchObject({
|
||||
id: 'run-1',
|
||||
status: 'queued',
|
||||
});
|
||||
expect(api.DataSyncRunStart).toHaveBeenCalledWith(
|
||||
task.id,
|
||||
task.revision,
|
||||
'one-time-token',
|
||||
);
|
||||
});
|
||||
|
||||
it('discards an unused production token when a fresh preflight changes the task', async () => {
|
||||
const task = taskFixture();
|
||||
const api = apiFixture();
|
||||
let clock = NOW;
|
||||
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => clock });
|
||||
const initialPreflight = await gateway.preflightTask(task);
|
||||
|
||||
await gateway.beginApproval(task, initialPreflight);
|
||||
clock = NOW + 10_000;
|
||||
await gateway.approveTask(task, initialPreflight);
|
||||
|
||||
const changed = reviseDataSyncTask(task, { name: 'changed after approval' });
|
||||
const refreshedPreflight = await gateway.preflightTask(changed);
|
||||
|
||||
await expect(gateway.startTask(changed, refreshedPreflight)).rejects.toThrow(
|
||||
'explicit production approval is required',
|
||||
);
|
||||
expect(api.DataSyncRunStart).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('binds authorization to the exact local signature and runnable lifecycle', async () => {
|
||||
const task = taskFixture();
|
||||
const api = apiFixture();
|
||||
@@ -343,7 +587,7 @@ describe('real Wails data sync gateway', () => {
|
||||
|
||||
const paused = reviseDataSyncTask(task, { lifecycle: 'paused' });
|
||||
await expect(gateway.startTask(paused, preflight)).rejects.toThrow(
|
||||
'only ready or enabled tasks can run',
|
||||
'preflight does not match the current task',
|
||||
);
|
||||
expect(api.DataSyncRunStart).not.toHaveBeenCalled();
|
||||
});
|
||||
@@ -366,6 +610,41 @@ describe('real Wails data sync gateway', () => {
|
||||
);
|
||||
});
|
||||
|
||||
it('sends the persisted revision when a schedule control pauses a loaded task', async () => {
|
||||
const task = { ...taskFixture(), lifecycle: 'enabled' as const };
|
||||
const paused = {
|
||||
...task,
|
||||
lifecycle: 'paused' as const,
|
||||
revision: task.revision + 1,
|
||||
};
|
||||
const api = apiFixture({
|
||||
DataSyncJobList: vi.fn(async () =>
|
||||
success([encodeDataSyncJobDefinition(task)]),
|
||||
),
|
||||
DataSyncJobSave: vi.fn(async (definition) =>
|
||||
success(encodeDataSyncJobDefinition(paused)),
|
||||
),
|
||||
});
|
||||
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => NOW });
|
||||
|
||||
const [loaded] = await gateway.listTasks();
|
||||
await expect(
|
||||
gateway.saveTask({ ...loaded, lifecycle: 'paused' }),
|
||||
).resolves.toMatchObject({
|
||||
lifecycle: 'paused',
|
||||
revision: paused.revision,
|
||||
});
|
||||
|
||||
expect(api.DataSyncJobSave).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
id: task.id,
|
||||
lifecycle: 'paused',
|
||||
revision: task.revision,
|
||||
}),
|
||||
'',
|
||||
);
|
||||
});
|
||||
|
||||
it('retries only a listed full-payload row at the current task revision', async () => {
|
||||
const task = taskFixture();
|
||||
const api = apiFixture({
|
||||
|
||||
@@ -6,6 +6,7 @@ import type {
|
||||
DataSyncApprovalChallenge,
|
||||
DataSyncApprovalGrant,
|
||||
DataSyncCheckpointSummary,
|
||||
DataSyncPreflightSnapshot,
|
||||
DataSyncRouteCapability,
|
||||
DataSyncRunRecord,
|
||||
DataSyncTaskDefinition,
|
||||
@@ -113,9 +114,11 @@ type GatewayOptions = {
|
||||
};
|
||||
|
||||
type CachedPreflight = {
|
||||
taskId: string;
|
||||
taskRevision: number;
|
||||
taskSignature: string;
|
||||
definitionHash: string;
|
||||
status: DataSyncPreflightSnapshot['status'];
|
||||
approvalRequired: boolean;
|
||||
canExecute: boolean;
|
||||
definition: WailsDataSyncJobDefinition;
|
||||
@@ -123,6 +126,8 @@ type CachedPreflight = {
|
||||
|
||||
type ApprovalToken = {
|
||||
token: string;
|
||||
taskId: string;
|
||||
taskRevision: number;
|
||||
expiresAt: string;
|
||||
definitionHash: string;
|
||||
taskSignature: string;
|
||||
@@ -130,6 +135,8 @@ type ApprovalToken = {
|
||||
|
||||
type ApprovalChallenge = {
|
||||
challenge: string;
|
||||
taskId: string;
|
||||
taskRevision: number;
|
||||
notBefore: string;
|
||||
expiresAt: string;
|
||||
definitionHash: string;
|
||||
@@ -189,6 +196,7 @@ const isCurrentPreflight = (
|
||||
): cached is CachedPreflight =>
|
||||
Boolean(
|
||||
cached &&
|
||||
cached.taskId === task.id &&
|
||||
cached.taskRevision === task.revision &&
|
||||
cached.taskSignature === taskSignature(task, previous),
|
||||
);
|
||||
@@ -214,6 +222,25 @@ export const createWailsDataSyncWorkbenchGateway = (
|
||||
return value.map((run) => decodeRunRecord(run, taskNames));
|
||||
};
|
||||
|
||||
const discardAuthorization = (taskId: string) => {
|
||||
approvalChallenges.delete(taskId);
|
||||
approvalTokens.delete(taskId);
|
||||
};
|
||||
|
||||
const requirePreflightTaskBinding = (
|
||||
task: DataSyncTaskDefinition,
|
||||
preflight: DataSyncPreflightSnapshot,
|
||||
operation: string,
|
||||
) => {
|
||||
if (preflight.taskId !== task.id || preflight.taskRevision !== task.revision) {
|
||||
discardAuthorization(task.id);
|
||||
throw new DataSyncGatewayProtocolError(
|
||||
operation,
|
||||
'preflight does not match the current task',
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
const takeApprovalToken = (
|
||||
task: DataSyncTaskDefinition,
|
||||
preflight: CachedPreflight,
|
||||
@@ -221,11 +248,13 @@ export const createWailsDataSyncWorkbenchGateway = (
|
||||
const approval = approvalTokens.get(task.id);
|
||||
if (
|
||||
!approval ||
|
||||
approval.taskId !== task.id ||
|
||||
approval.taskRevision !== task.revision ||
|
||||
approval.definitionHash !== preflight.definitionHash ||
|
||||
approval.taskSignature !== preflight.taskSignature ||
|
||||
Date.parse(approval.expiresAt) <= now()
|
||||
) {
|
||||
approvalTokens.delete(task.id);
|
||||
discardAuthorization(task.id);
|
||||
return '';
|
||||
}
|
||||
// The backend token is one-time. Remove it before crossing the boundary so
|
||||
@@ -240,6 +269,7 @@ export const createWailsDataSyncWorkbenchGateway = (
|
||||
const previous = wireJobs.get(task.id);
|
||||
const cached = preflights.get(task.id);
|
||||
if (!isCurrentPreflight(cached, task, previous)) {
|
||||
discardAuthorization(task.id);
|
||||
throw new DataSyncGatewayProtocolError(
|
||||
'data sync preflight',
|
||||
'task definition changed; run preflight again',
|
||||
@@ -248,6 +278,14 @@ export const createWailsDataSyncWorkbenchGateway = (
|
||||
return cached;
|
||||
};
|
||||
|
||||
const preflightMatchesCache = (
|
||||
cached: CachedPreflight,
|
||||
preflight: DataSyncPreflightSnapshot,
|
||||
): boolean =>
|
||||
cached.status === preflight.status &&
|
||||
cached.definitionHash === preflight.definitionHash &&
|
||||
cached.approvalRequired === preflight.approvalRequired;
|
||||
|
||||
const getCheckpoint = async (
|
||||
taskId: string,
|
||||
): Promise<DataSyncCheckpointSummary | null> => {
|
||||
@@ -418,6 +456,7 @@ export const createWailsDataSyncWorkbenchGateway = (
|
||||
}
|
||||
const previous = wireJobs.get(task.id);
|
||||
const input = encodeDataSyncJobDefinition(task, previous);
|
||||
const inputSignature = JSON.stringify(input);
|
||||
const decoded = decodeDataSyncPreflightQuery(
|
||||
await api.DataSyncJobPreflight(asJobDefinition(input)),
|
||||
task,
|
||||
@@ -433,12 +472,27 @@ export const createWailsDataSyncWorkbenchGateway = (
|
||||
});
|
||||
}
|
||||
const definition = sanitizedWireDefinition(decoded.definition);
|
||||
const approved = approvalTokens.get(task.id);
|
||||
approvalChallenges.delete(task.id);
|
||||
approvalTokens.delete(task.id);
|
||||
if (
|
||||
!approved ||
|
||||
approved.taskId !== task.id ||
|
||||
approved.taskRevision !== task.revision ||
|
||||
decoded.snapshot.status === 'blocked' ||
|
||||
!decoded.snapshot.approvalRequired ||
|
||||
!decoded.capability.canExecute ||
|
||||
approved.definitionHash !== decoded.snapshot.definitionHash ||
|
||||
approved.taskSignature !== inputSignature ||
|
||||
Date.parse(approved.expiresAt) <= now()
|
||||
) {
|
||||
approvalTokens.delete(task.id);
|
||||
}
|
||||
preflights.set(task.id, {
|
||||
taskId: task.id,
|
||||
taskRevision: task.revision,
|
||||
taskSignature: JSON.stringify(input),
|
||||
taskSignature: inputSignature,
|
||||
definitionHash: decoded.snapshot.definitionHash,
|
||||
status: decoded.snapshot.status,
|
||||
approvalRequired: decoded.snapshot.approvalRequired,
|
||||
canExecute: decoded.capability.canExecute,
|
||||
definition,
|
||||
@@ -447,12 +501,15 @@ export const createWailsDataSyncWorkbenchGateway = (
|
||||
},
|
||||
|
||||
async beginApproval(task, preflight): Promise<DataSyncApprovalChallenge> {
|
||||
requirePreflightTaskBinding(task, preflight, 'DataSyncJobApprovalBegin');
|
||||
const cached = requireCurrentPreflight(task);
|
||||
if (
|
||||
!preflight.approvalRequired ||
|
||||
preflight.status === 'blocked' ||
|
||||
cached.definitionHash !== preflight.definitionHash
|
||||
!preflightMatchesCache(cached, preflight) ||
|
||||
!cached.canExecute
|
||||
) {
|
||||
discardAuthorization(task.id);
|
||||
throw new DataSyncGatewayProtocolError(
|
||||
'DataSyncJobApprovalBegin',
|
||||
'approval does not match the current passed preflight',
|
||||
@@ -472,28 +529,39 @@ export const createWailsDataSyncWorkbenchGateway = (
|
||||
}
|
||||
approvalChallenges.set(task.id, {
|
||||
...challenge,
|
||||
taskId: task.id,
|
||||
taskRevision: cached.taskRevision,
|
||||
definitionHash: preflight.definitionHash,
|
||||
taskSignature: cached.taskSignature,
|
||||
});
|
||||
return {
|
||||
taskId: task.id,
|
||||
definitionHash: preflight.definitionHash,
|
||||
taskRevision: cached.taskRevision,
|
||||
notBefore: challenge.notBefore,
|
||||
expiresAt: challenge.expiresAt,
|
||||
};
|
||||
},
|
||||
|
||||
async approveTask(task, preflight): Promise<DataSyncApprovalGrant> {
|
||||
requirePreflightTaskBinding(task, preflight, 'DataSyncJobApprove');
|
||||
const cached = requireCurrentPreflight(task);
|
||||
const challenge = approvalChallenges.get(task.id);
|
||||
if (
|
||||
!challenge ||
|
||||
challenge.taskId !== task.id ||
|
||||
challenge.taskRevision !== task.revision ||
|
||||
!preflightMatchesCache(cached, preflight) ||
|
||||
cached.status === 'blocked' ||
|
||||
!cached.approvalRequired ||
|
||||
!cached.canExecute ||
|
||||
challenge.definitionHash !== preflight.definitionHash ||
|
||||
challenge.definitionHash !== cached.definitionHash ||
|
||||
challenge.taskSignature !== cached.taskSignature ||
|
||||
Date.parse(challenge.notBefore) > now() ||
|
||||
Date.parse(challenge.expiresAt) <= now()
|
||||
) {
|
||||
approvalChallenges.delete(task.id);
|
||||
discardAuthorization(task.id);
|
||||
throw new DataSyncGatewayProtocolError(
|
||||
'DataSyncJobApprove',
|
||||
'backend approval countdown is incomplete, expired, or stale',
|
||||
@@ -519,16 +587,21 @@ export const createWailsDataSyncWorkbenchGateway = (
|
||||
}
|
||||
approvalTokens.set(task.id, {
|
||||
...approved,
|
||||
taskId: task.id,
|
||||
taskRevision: cached.taskRevision,
|
||||
definitionHash: preflight.definitionHash,
|
||||
taskSignature: cached.taskSignature,
|
||||
});
|
||||
return {
|
||||
taskId: task.id,
|
||||
definitionHash: preflight.definitionHash,
|
||||
taskRevision: cached.taskRevision,
|
||||
expiresAt: approved.expiresAt,
|
||||
};
|
||||
},
|
||||
|
||||
async startTask(task, preflight) {
|
||||
requirePreflightTaskBinding(task, preflight, 'DataSyncRunStart');
|
||||
if (isLocalDataSyncTaskId(task.id)) {
|
||||
throw new DataSyncGatewayProtocolError(
|
||||
'DataSyncRunStart',
|
||||
@@ -543,10 +616,11 @@ export const createWailsDataSyncWorkbenchGateway = (
|
||||
}
|
||||
const cached = requireCurrentPreflight(task);
|
||||
if (
|
||||
cached.definitionHash !== preflight.definitionHash ||
|
||||
preflight.status === 'blocked' ||
|
||||
!preflightMatchesCache(cached, preflight) ||
|
||||
cached.status === 'blocked' ||
|
||||
!cached.canExecute
|
||||
) {
|
||||
discardAuthorization(task.id);
|
||||
throw new DataSyncGatewayProtocolError(
|
||||
'DataSyncRunStart',
|
||||
'preflight is blocked or stale',
|
||||
|
||||
@@ -231,6 +231,121 @@ func TestStorePersistsIncompleteDraftButManagerWillNotRunIt(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestStorePersistsPausedScheduledJobAcrossReopen(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
path := t.TempDir() + "/sync-jobs.db"
|
||||
store, err := Open(path)
|
||||
if err != nil {
|
||||
t.Fatalf("open store: %v", err)
|
||||
}
|
||||
|
||||
definition, err := store.PutJob(ctx, JobDefinition{
|
||||
Name: "scheduled orders sync",
|
||||
Lifecycle: JobLifecycleEnabled,
|
||||
Kind: JobKindReconcile,
|
||||
IncrementalMode: IncrementalSnapshot,
|
||||
Source: EndpointRef{ConnectionID: "source"},
|
||||
Target: EndpointRef{ConnectionID: "target"},
|
||||
Mappings: []TableMapping{{
|
||||
SourceTable: "orders",
|
||||
TargetTable: "orders_archive",
|
||||
Enabled: true,
|
||||
}},
|
||||
Schedule: ScheduleSpec{Kind: ScheduleInterval, IntervalSeconds: 60},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("save scheduled job: %v", err)
|
||||
}
|
||||
if definition.NextRunAt == 0 {
|
||||
t.Fatal("enabled scheduled job has no next run")
|
||||
}
|
||||
|
||||
paused, err := store.PauseJob(ctx, definition.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("pause scheduled job: %v", err)
|
||||
}
|
||||
if paused.Lifecycle != JobLifecyclePaused || paused.Enabled || paused.NextRunAt != 0 {
|
||||
t.Fatalf("paused job = %#v", paused)
|
||||
}
|
||||
if err := store.Close(); err != nil {
|
||||
t.Fatalf("close store: %v", err)
|
||||
}
|
||||
|
||||
reopened, err := Open(path)
|
||||
if err != nil {
|
||||
t.Fatalf("reopen store: %v", err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
if err := reopened.Close(); err != nil {
|
||||
t.Errorf("close reopened store: %v", err)
|
||||
}
|
||||
})
|
||||
|
||||
persisted, err := reopened.GetJob(ctx, definition.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("get paused job after reopen: %v", err)
|
||||
}
|
||||
if persisted.Lifecycle != JobLifecyclePaused || persisted.Enabled || persisted.NextRunAt != 0 {
|
||||
t.Fatalf("persisted paused job = %#v", persisted)
|
||||
}
|
||||
due, err := reopened.ListDueJobs(ctx, time.Now().Add(24*time.Hour).UnixMilli())
|
||||
if err != nil {
|
||||
t.Fatalf("list due jobs after reopen: %v", err)
|
||||
}
|
||||
if len(due) != 0 {
|
||||
t.Fatalf("paused job returned as due after reopen: %#v", due)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStorePausedPutCancelsQueuedRunsAndRequestsRunningCancellation(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := openTestStore(t)
|
||||
definition := putTestJob(t, store, "queue")
|
||||
snapshot, err := json.Marshal(definition)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal definition snapshot: %v", err)
|
||||
}
|
||||
queued, err := store.CreateRun(ctx, RunRecord{
|
||||
JobID: definition.ID, JobRevision: definition.Revision, Status: RunStatusQueued,
|
||||
DefinitionSnapshot: snapshot,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("create queued run: %v", err)
|
||||
}
|
||||
running, err := store.CreateRun(ctx, RunRecord{
|
||||
JobID: definition.ID, JobRevision: definition.Revision, Status: RunStatusRunning,
|
||||
DefinitionSnapshot: snapshot,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("create running run: %v", err)
|
||||
}
|
||||
|
||||
definition.Lifecycle = JobLifecyclePaused
|
||||
definition.Enabled = false
|
||||
paused, err := store.PutJob(ctx, definition)
|
||||
if err != nil {
|
||||
t.Fatalf("save paused job: %v", err)
|
||||
}
|
||||
if paused.Lifecycle != JobLifecyclePaused || paused.Enabled {
|
||||
t.Fatalf("paused job = %#v", paused)
|
||||
}
|
||||
|
||||
persistedQueued, err := store.GetRun(ctx, queued.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("get queued run: %v", err)
|
||||
}
|
||||
if persistedQueued.Status != RunStatusCanceled || persistedQueued.FinishedAt == 0 || persistedQueued.Message != "canceled because task was paused" {
|
||||
t.Fatalf("queued run after pause = %#v", persistedQueued)
|
||||
}
|
||||
persistedRunning, err := store.GetRun(ctx, running.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("get running run: %v", err)
|
||||
}
|
||||
if persistedRunning.Status != RunStatusCancelling || persistedRunning.FinishedAt != 0 || persistedRunning.Message != "cancellation requested because task was paused" {
|
||||
t.Fatalf("running run after pause = %#v", persistedRunning)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStoreResetCheckpointRejectsActiveRun(t *testing.T) {
|
||||
store := openTestStore(t)
|
||||
definition := putTestJob(t, store, "forbid")
|
||||
|
||||
Reference in New Issue
Block a user