fix(transfer): confirm background task submission

Closes #8649
This commit is contained in:
t8y2
2026-09-27 14:19:10 +00:00
parent 5d13c5ecfc
commit 4329ec51db
20 changed files with 181 additions and 12 deletions
@@ -4,7 +4,7 @@ import { uuid } from "@/lib/common/utils";
import { useI18n } from "vue-i18n";
import { Dialog, DialogContent, DialogDescription, DialogHeader, DialogTitle, DialogFooter } from "@/components/ui/dialog";
import { Button } from "@/components/ui/button";
import { buildTransferObjectSelections } from "./transferSelections";
import { buildTransferObjectSelections, countTransferObjects } from "./transferSelections";
import { createTaskLoadTracker } from "./taskLoadTracker";
import { confirmTransferWithProductionSafety, createTransferSubmission, rebuildUnavailableReason, resolveTransferStrategy, transferStrategyOptions, type TransferStrategy } from "./transferStrategy";
import { Input } from "@/components/ui/input";
@@ -815,6 +815,7 @@ function runTransfer(request: api.TransferRequest, shouldRefreshTargetTree: bool
isSubmitting.value = true;
startDataTransferTask(request, `${request.sourceDatabase} → ${request.targetDatabase}`, {
formatOverlapError: (tables) => t("transfer.targetTableBusy", { tables: tables.join(", ") }),
onStarted: () => toast(t("transfer.backgroundStarted")),
onDone: async () => {
if (shouldRefreshTargetTree) {
await store.refreshObjectListTreeNode(request.targetConnectionId, request.targetDatabase, request.targetSchema, request.targetCatalog);
@@ -1052,7 +1053,7 @@ const confirmationSummary = computed(() => {
if (!request) return "";
const source = formatTransferEndpointLabel(getConnectionName(request.sourceConnectionId), request.sourceDatabase, request.sourceSchema, request.sourceCatalog);
const target = formatTransferEndpointLabel(getConnectionName(request.targetConnectionId), request.targetDatabase, request.targetSchema, request.targetCatalog);
const count = request.tables.length + request.objects.reduce((total, selection) => total + selection.names.length, 0);
const count = countTransferObjects(request);
return t("transfer.startConfirmMessage", { source, target, count });
});
@@ -1,10 +1,28 @@
import { describe, expect, it } from "vitest";
import { buildTransferObjectSelections, matchBulkObjectNames, parseBulkObjectNames } from "../transferSelections";
import { buildTransferObjectSelections, countTransferObjects, matchBulkObjectNames, parseBulkObjectNames } from "../transferSelections";
function setOf(names: string[]): Set<string> {
return new Set(names);
}
describe("countTransferObjects", () => {
it("counts selected tables and non-table objects from their separate request fields", () => {
expect(
countTransferObjects({
tables: ["orders", "users"],
objects: [
{ objectType: "VIEW", names: ["active_users"] },
{ objectType: "SEQUENCE", names: ["order_id_seq"] },
],
}),
).toBe(4);
});
it("keeps a table-only multi-selection nonzero", () => {
expect(countTransferObjects({ tables: ["orders", "users"], objects: [] })).toBe(2);
});
});
describe("buildTransferObjectSelections", () => {
it("serializes non-table selections in request order", () => {
const result = buildTransferObjectSelections(
@@ -1,10 +1,15 @@
import type { TransferObjectKind } from "@/lib/backend/api";
import type { TransferObjectKind, TransferRequest } from "@/lib/backend/api";
export interface TransferObjectSelectionPayload {
objectType: TransferObjectKind;
names: string[];
}
/** Counts table and non-table selections represented by a transfer request. */
export function countTransferObjects(request: Pick<TransferRequest, "tables" | "objects">): number {
return request.tables.length + request.objects.reduce((total, selection) => total + selection.names.length, 0);
}
/**
* Builds the `objects` payload of a transfer request from the tree
* selection state. TABLE is handled by `tables`, and object types that
@@ -35,10 +35,14 @@ function transferRequest(transferId: string, tables = ["users"]): TransferReques
targetSchema: "public",
tables,
createTable: true,
content: "structureAndData",
objects: [],
mode: "append",
targetTableNameCase: "preserve",
quoteTargetColumnNames: true,
batchSize: 1000,
dropTargetBeforeCreate: false,
dropTargetConfirmed: false,
};
}
@@ -138,16 +142,46 @@ describe("data transfer task duration", () => {
it("records an immediate start failure as a terminal duration", async () => {
vi.mocked(api.startTransfer).mockRejectedValueOnce(new Error("start failed"));
const tracker = useExportTracker();
const onStarted = vi.fn();
now = 10_000;
const task = tracker.startDataTransferTask(transferRequest("start-failure"), "users");
const task = tracker.startDataTransferTask(transferRequest("start-failure"), "users", { onStarted });
now = 10_025;
await vi.waitFor(() => expect(task.status).toBe("Error"));
expect(onStarted).not.toHaveBeenCalled();
expect(task.finishedAt! - task.startedAt!).toBe(25);
expect(task.errorMessage).toBe("start failed");
});
it.each([
["append", { content: "structureAndData", mode: "append", createTable: true, dropTargetBeforeCreate: false }],
["rebuild", { content: "structureAndData", mode: "append", createTable: true, dropTargetBeforeCreate: true }],
["data-only", { content: "dataOnly", mode: "append", createTable: false, dropTargetBeforeCreate: false }],
] as const)("launches one %s task and acknowledges its accepted submission once", async (_flow, overrides) => {
let finishTransfer!: () => void;
vi.mocked(api.startTransfer).mockImplementationOnce((_request, _onProgress, onStarted) => {
onStarted?.();
onStarted?.();
return new Promise<void>((resolve) => {
finishTransfer = resolve;
});
});
const tracker = useExportTracker();
const onStarted = vi.fn();
const request = { ...transferRequest(`start-${_flow}`), ...overrides };
const task = tracker.startDataTransferTask(request, _flow, { onStarted });
expect(task.status).toBe("Running");
expect(api.startTransfer).toHaveBeenCalledTimes(1);
expect(vi.mocked(api.startTransfer).mock.calls[0]?.[0]).toMatchObject(overrides);
expect(onStarted).toHaveBeenCalledTimes(1);
finishTransfer();
await Promise.resolve();
});
it("freezes an overlapping transfer failure without starting another request", async () => {
let resolveFirst!: () => void;
vi.mocked(api.startTransfer).mockImplementationOnce(() => new Promise<void>((resolve) => (resolveFirst = resolve)));
@@ -155,11 +189,13 @@ describe("data transfer task duration", () => {
now = 100;
tracker.startDataTransferTask(transferRequest("active"), "active");
now = 130;
const overlapping = tracker.startDataTransferTask(transferRequest("overlap"), "overlap");
const onStarted = vi.fn();
const overlapping = tracker.startDataTransferTask(transferRequest("overlap"), "overlap", { onStarted });
expect(overlapping.status).toBe("Error");
expect(overlapping.finishedAt! - overlapping.startedAt!).toBe(0);
expect(api.startTransfer).toHaveBeenCalledTimes(1);
expect(onStarted).not.toHaveBeenCalled();
resolveFirst();
await Promise.resolve();
@@ -582,6 +582,7 @@ export function useExportTracker() {
request: api.TransferRequest,
label: string,
options: {
onStarted?: () => void;
onDone?: () => void | Promise<void>;
formatOverlapError?: (tables: string[]) => string;
} = {},
@@ -607,13 +608,23 @@ export function useExportTracker() {
activeTransferRuns.add(request.transferId);
let terminalStatus: api.TransferProgress["status"] | null = null;
let startAcknowledged = false;
const acknowledgeStart = () => {
if (startAcknowledged) return;
startAcknowledged = true;
options.onStarted?.();
};
void (async () => {
try {
await api.startTransfer(request, (progress) => {
terminalStatus = isTerminalTransferProgress(progress) ? progress.status : terminalStatus;
updateDataTransferTask(progress.transferId, progress);
});
await api.startTransfer(
request,
(progress) => {
terminalStatus = isTerminalTransferProgress(progress) ? progress.status : terminalStatus;
updateDataTransferTask(progress.transferId, progress);
},
acknowledgeStart,
);
if (terminalStatus === "done" && task.status === "Done") {
await options.onDone?.();
@@ -67,6 +67,7 @@ const FEATURE_MESSAGES: FeatureMessages[] = [
{ feature: "plugin marketplace sorting (#10078)", keys: under("pluginPlatform", ["sortBy", "sortByName", "sortByRecentlyUpdated", "sortByRecentlyListed", "sortByUpdatesFirst"]), locales: EXCEPT_AZ_TR, translated: true },
{ feature: "shared refresh action (#8768)", keys: ["common.refresh"] },
{ feature: "transfer bulk select", keys: under("transfer", ["bulkSelectObjects", "bulkSelectTitle", "bulkSelectHint", "bulkSelectPlaceholder", "bulkSelectConfirm", "bulkSelectMatched", "bulkSelectUnmatched", "noMatchingObjects"]), translated: true },
{ feature: "transfer background acknowledgement (#8649)", keys: ["transfer.backgroundStarted"], translated: true },
{ feature: "table info pin", keys: ["grid.pinTableInfo", "grid.unpinTableInfo"], translated: true },
{ feature: "Consul workspace", keys: ["consul.ui.*", "consul.tools.*"] },
{ feature: "MQTT max packet size", keys: ["connection.mqttMaxPacketSize"] },
+1
View File
@@ -6159,6 +6159,7 @@ export default withEnglishFallback({
overallProgress: "Ümumi gedişat",
dataTransfer: "Məlumatların köçürülməsi",
runInBackground: "Arxa planda işlət",
backgroundStarted: "Köçürmə arxa planda başladı. Gedişatı fon tapşırıqlarında izləyə bilərsiniz.",
newTask: "Yeni köçürmə",
editConfig: "Konfiqurasiyanı redaktə et",
retry: "Yenidən cəhd et",
+1
View File
@@ -6760,6 +6760,7 @@ export default {
overallProgress: "Overall progress",
dataTransfer: "Data Transfer",
runInBackground: "Run in background",
backgroundStarted: "Transfer started in the background. View progress in Background Tasks.",
newTask: "New transfer",
editConfig: "Edit config",
retry: "Retry",
+1
View File
@@ -6198,6 +6198,7 @@ export default withEnglishFallback({
swap: "Intercambiar origen y destino",
targetSchema: "Esquema de destino",
runInBackground: "Ejecutar en segundo plano",
backgroundStarted: "La transferencia comenzó en segundo plano. Consulta el progreso en Tareas en segundo plano.",
newTask: "Nueva transferencia",
editConfig: "Editar config",
retry: "Reintentar",
+1
View File
@@ -6121,6 +6121,7 @@ export default withEnglishFallback({
overallProgress: "Progresso complessivo",
dataTransfer: "Trasferimento Dati",
runInBackground: "Esegui in background",
backgroundStarted: "Il trasferimento è iniziato in background. Puoi verificarne l'avanzamento nelle attività in background.",
newTask: "Nuovo trasferimento",
editConfig: "Modifica config",
retry: "Riprova",
+1
View File
@@ -6142,6 +6142,7 @@ export default withEnglishFallback({
overallProgress: "全体の進捗",
dataTransfer: "データ転送",
runInBackground: "バックグラウンドで実行",
backgroundStarted: "転送をバックグラウンドで開始しました。進行状況はバックグラウンドタスクで確認できます。",
newTask: "新規転送",
editConfig: "設定を編集",
retry: "リトライ",
+1
View File
@@ -5848,6 +5848,7 @@ export default withEnglishFallback({
overallProgress: "전체 진행률",
dataTransfer: "데이터 이전",
runInBackground: "백그라운드에서 실행",
backgroundStarted: "전송이 백그라운드에서 시작되었습니다. 백그라운드 작업에서 진행 상황을 확인할 수 있습니다.",
newTask: "새 이전",
editConfig: "설정 편집",
retry: "다시 시도",
+1
View File
@@ -6122,6 +6122,7 @@ export default withEnglishFallback({
overallProgress: "Progresso geral",
dataTransfer: "Transferência de Dados",
runInBackground: "Executar em segundo plano",
backgroundStarted: "A transferência foi iniciada em segundo plano. Acompanhe o progresso nas tarefas em segundo plano.",
newTask: "Nova transferência",
editConfig: "Editar configuração",
retry: "Tentar novamente",
+1
View File
@@ -7280,6 +7280,7 @@ export default withEnglishFallback({
overallProgress: "Общий прогресс",
dataTransfer: "Перенос данных",
runInBackground: "Выполнить в фоне",
backgroundStarted: "Перенос запущен в фоновом режиме. Ход выполнения можно посмотреть в фоновых задачах.",
newTask: "Новый перенос",
editConfig: "Изменить конфигурацию",
retry: "Повторить",
+1
View File
@@ -6043,6 +6043,7 @@ export default withEnglishFallback({
overallProgress: "Genel ilerleme",
dataTransfer: "Veri Aktarımı",
runInBackground: "Arka planda çalıştır",
backgroundStarted: "Aktarım arka planda başlatıldı. İlerlemeyi arka plan görevlerinde görüntüleyebilirsiniz.",
newTask: "Yeni aktarım",
editConfig: "Yapılandırmayı düzenle",
retry: "Yeniden dene",
+1
View File
@@ -6727,6 +6727,7 @@ export default withEnglishFallback({
overallProgress: "整体进度",
dataTransfer: "数据传输",
runInBackground: "后台运行",
backgroundStarted: "传输已在后台开始,可在后台任务中查看进度。",
newTask: "新建传输",
editConfig: "编辑配置",
retry: "重新传输",
+1
View File
@@ -5431,6 +5431,7 @@ export default withEnglishFallback({
tableNameCaseLower: "小寫",
tableNameCaseUpper: "大寫",
runInBackground: "背景執行",
backgroundStarted: "傳輸已在背景開始,可在背景任務中查看進度。",
newTask: "新增傳輸",
editConfig: "編輯配置",
retry: "重試",
@@ -0,0 +1,84 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import type { TransferProgress, TransferRequest } from "@/lib/backend/http";
import { startTransfer } from "@/lib/backend/http";
function request(): TransferRequest {
return {
transferId: "transfer-1",
sourceConnectionId: "source",
sourceDatabase: "app",
sourceSchema: "public",
targetConnectionId: "target",
targetDatabase: "warehouse",
targetSchema: "reporting",
tables: ["orders", "users"],
createTable: true,
content: "structureAndData",
objects: [],
mode: "append",
targetTableNameCase: "preserve",
quoteTargetColumnNames: true,
ownershipPolicy: "preserve",
batchSize: 1000,
dropTargetBeforeCreate: false,
dropTargetConfirmed: false,
};
}
function doneProgress(): TransferProgress {
return {
transferId: "transfer-1",
table: "users",
tableIndex: 2,
totalTables: 2,
rowsTransferred: 3,
totalRows: 3,
status: "done",
error: null,
terminal: true,
};
}
describe("web data-transfer submission", () => {
afterEach(() => vi.unstubAllGlobals());
it("acknowledges an accepted task once before tracking its progress", async () => {
vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true }));
let source: FakeEventSource | undefined;
class FakeEventSource {
onmessage: ((event: { data: string }) => void) | null = null;
onerror: (() => void) | null = null;
close = vi.fn();
constructor(readonly url: string) {
source = this;
}
}
vi.stubGlobal("EventSource", FakeEventSource);
const onStarted = vi.fn();
const onProgress = vi.fn();
const pending = startTransfer(request(), onProgress, onStarted);
await vi.waitFor(() => expect(onStarted).toHaveBeenCalledTimes(1));
expect(source?.url).toBe("/api/transfer/progress/transfer-1");
source?.onmessage?.({ data: JSON.stringify(doneProgress()) });
await expect(pending).resolves.toBeUndefined();
expect(onStarted).toHaveBeenCalledTimes(1);
expect(onProgress).toHaveBeenCalledWith(doneProgress());
});
it("suppresses acknowledgement when task creation fails", async () => {
vi.stubGlobal(
"fetch",
vi.fn().mockResolvedValue({
ok: false,
text: vi.fn().mockResolvedValue('{"message":"rejected"}'),
}),
);
const onStarted = vi.fn();
await expect(startTransfer(request(), vi.fn(), onStarted)).rejects.toThrow();
expect(onStarted).not.toHaveBeenCalled();
});
});
+2 -1
View File
@@ -2805,7 +2805,7 @@ export async function saveGlobalSearchSettings(settings: GlobalSearchSettings):
// Data Transfer
// ---------------------------------------------------------------------------
export async function startTransfer(request: TransferRequest, onProgress: (progress: TransferProgress) => void): Promise<void> {
export async function startTransfer(request: TransferRequest, onProgress: (progress: TransferProgress) => void, onStarted?: () => void): Promise<void> {
// 1. POST to start the transfer
const res = await fetch(apiUrl("/api/transfer/start"), {
method: "POST",
@@ -2813,6 +2813,7 @@ export async function startTransfer(request: TransferRequest, onProgress: (progr
body: JSON.stringify({ request }),
});
if (!res.ok) throw await backendResponseError(res);
onStarted?.();
// 2. SSE to listen for progress
return new Promise((resolve, reject) => {
+2 -1
View File
@@ -5324,7 +5324,7 @@ export interface TransferProgress {
transferFailuresOmitted?: number;
}
export async function startTransfer(request: TransferRequest, onProgress: (progress: TransferProgress) => void): Promise<void> {
export async function startTransfer(request: TransferRequest, onProgress: (progress: TransferProgress) => void, onStarted?: () => void): Promise<void> {
return new Promise((resolve, reject) => {
let unlisten: UnlistenFn | null = null;
void (async () => {
@@ -5339,6 +5339,7 @@ export async function startTransfer(request: TransferRequest, onProgress: (progr
});
await invoke("start_transfer", { request });
onStarted?.();
} catch (e) {
unlisten?.();
reject(e instanceof BackendErrorException ? e : new BackendErrorException(e));