From 15d392bc44539309c734c7f992bd5b52b327b4f9 Mon Sep 17 00:00:00 2001 From: Guoyu Su Date: Thu, 9 Jul 2026 18:30:29 +0800 Subject: [PATCH] fix(transfer): keep task active after table errors --- .../src/composables/useExportTracker.ts | 10 +-- apps/desktop/src/lib/backend/http.ts | 3 +- apps/desktop/src/lib/backend/tauri.ts | 4 +- .../src/lib/backend/transferProgress.ts | 8 +++ crates/dbx-core/src/transfer.rs | 13 ++++ crates/dbx-web/src/routes/transfer.rs | 9 +++ packages/app-tests/transferProgress.test.ts | 12 ++++ packages/app-tests/useExportTracker.test.ts | 62 +++++++++++++++++-- src-tauri/src/commands/transfer.rs | 9 +++ 9 files changed, 119 insertions(+), 11 deletions(-) create mode 100644 apps/desktop/src/lib/backend/transferProgress.ts create mode 100644 packages/app-tests/transferProgress.test.ts diff --git a/apps/desktop/src/composables/useExportTracker.ts b/apps/desktop/src/composables/useExportTracker.ts index d41a91f69c..0313b3428e 100644 --- a/apps/desktop/src/composables/useExportTracker.ts +++ b/apps/desktop/src/composables/useExportTracker.ts @@ -1,5 +1,6 @@ import { reactive, computed } from "vue"; import * as api from "@/lib/backend/api"; +import { isTerminalTransferProgress } from "@/lib/backend/transferProgress"; export type BackgroundTaskKind = "table-export" | "database-export" | "sql-file" | "data-transfer"; export type BackgroundTaskStatus = "Running" | "Writing" | "Done" | "Error" | "Cancelled"; @@ -46,9 +47,9 @@ function normalizeSqlFileStatus(status: api.SqlFileStatus): BackgroundTaskStatus return "Running"; } -function normalizeTransferStatus(status: api.TransferProgress["status"]): BackgroundTaskStatus { +function normalizeTransferStatus(status: api.TransferProgress["status"], terminal: boolean): BackgroundTaskStatus { if (status === "done") return "Done"; - if (status === "error") return "Error"; + if (status === "error") return terminal ? "Error" : "Running"; if (status === "cancelled") return "Cancelled"; return "Running"; } @@ -200,7 +201,7 @@ export function useExportTracker() { void (async () => { try { await api.startTransfer(request, (progress) => { - terminalStatus = progress.status === "done" || progress.status === "error" || progress.status === "cancelled" ? progress.status : terminalStatus; + terminalStatus = isTerminalTransferProgress(progress) ? progress.status : terminalStatus; updateDataTransferTask(progress.transferId, progress); }); @@ -217,6 +218,7 @@ export function useExportTracker() { totalRows: task.totalRows, status: "error", error: e?.message || String(e), + terminal: true, }); } finally { activeTransferRuns.delete(request.transferId); @@ -266,7 +268,7 @@ export function useExportTracker() { function updateDataTransferTask(transferId: string, progress: api.TransferProgress) { const task = taskMap.get(transferId); if (!task) return; - const nextStatus = normalizeTransferStatus(progress.status); + const nextStatus = normalizeTransferStatus(progress.status, progress.terminal); const hadError = task.status === "Error"; task.status = hadError && nextStatus === "Done" ? "Error" : nextStatus; task.errorMessage = progress.error || task.errorMessage || null; diff --git a/apps/desktop/src/lib/backend/http.ts b/apps/desktop/src/lib/backend/http.ts index 39f35d075a..d741e864ce 100644 --- a/apps/desktop/src/lib/backend/http.ts +++ b/apps/desktop/src/lib/backend/http.ts @@ -100,6 +100,7 @@ import type { AppSupportInfo, } from "@/lib/backend/tauri"; import type { QueryEditability } from "@/lib/sql/sqlAnalysis"; +import { isTerminalTransferProgress } from "@/lib/backend/transferProgress"; import type { DataGridColumnDistinctValuesSqlOptions, DataGridColumnValueFilterConditionOptions, @@ -1316,7 +1317,7 @@ export async function startTransfer(request: TransferRequest, onProgress: (progr es.onmessage = (e) => { const progress: TransferProgress = JSON.parse(e.data); onProgress(progress); - if (progress.status === "done" || progress.status === "error" || progress.status === "cancelled") { + if (isTerminalTransferProgress(progress)) { es.close(); resolve(); } diff --git a/apps/desktop/src/lib/backend/tauri.ts b/apps/desktop/src/lib/backend/tauri.ts index 6569683031..42d6353d70 100644 --- a/apps/desktop/src/lib/backend/tauri.ts +++ b/apps/desktop/src/lib/backend/tauri.ts @@ -38,6 +38,7 @@ import type { CollectionInfo } from "@/types/database"; import type { SidebarObjectKind } from "@/lib/database/databaseObjectCapabilities"; import type { AiConfig, AiTestConnectionResult } from "@/stores/settingsStore"; import type { QueryEditability } from "@/lib/sql/sqlAnalysis"; +import { isTerminalTransferProgress } from "@/lib/backend/transferProgress"; import type { DataGridColumnDistinctValuesSqlOptions, DataGridColumnValueFilterConditionOptions, @@ -1942,6 +1943,7 @@ export interface TransferProgress { totalRows: number | null; status: "running" | "tableDone" | "done" | "error" | "cancelled"; error: string | null; + terminal: boolean; } export async function startTransfer(request: TransferRequest, onProgress: (progress: TransferProgress) => void): Promise { @@ -1952,7 +1954,7 @@ export async function startTransfer(request: TransferRequest, onProgress: (progr unlisten = await listen("transfer-progress", (event) => { if (event.payload.transferId !== request.transferId) return; onProgress(event.payload); - if (event.payload.status === "done" || event.payload.status === "error" || event.payload.status === "cancelled") { + if (isTerminalTransferProgress(event.payload)) { unlisten?.(); resolve(); } diff --git a/apps/desktop/src/lib/backend/transferProgress.ts b/apps/desktop/src/lib/backend/transferProgress.ts new file mode 100644 index 0000000000..a190fcf983 --- /dev/null +++ b/apps/desktop/src/lib/backend/transferProgress.ts @@ -0,0 +1,8 @@ +type TransferProgressTerminalState = { + terminal?: boolean; + status: "running" | "tableDone" | "done" | "error" | "cancelled"; +}; + +export function isTerminalTransferProgress(progress: TransferProgressTerminalState): boolean { + return progress.terminal === true || progress.status === "done" || progress.status === "cancelled"; +} diff --git a/crates/dbx-core/src/transfer.rs b/crates/dbx-core/src/transfer.rs index afed6e52d6..065a93d5df 100644 --- a/crates/dbx-core/src/transfer.rs +++ b/crates/dbx-core/src/transfer.rs @@ -99,6 +99,7 @@ pub struct TransferProgress { pub total_rows: Option, pub status: TransferStatus, pub error: Option, + pub terminal: bool, } #[derive(Debug, Clone, Serialize)] @@ -3789,6 +3790,7 @@ where total_rows, status: TransferStatus::Running, error: None, + terminal: false, }); if row_count < batch_size { @@ -4148,6 +4150,7 @@ where total_rows, status: TransferStatus::Running, error: None, + terminal: false, }); if row_count < batch_size { @@ -4232,6 +4235,7 @@ where total_rows: Some(total_steps as u64), status: TransferStatus::Running, error: None, + terminal: false, }); execute_on_pool(state, target_pool_key, &generate_postgres_extension_ddl(&extension, &request.target_schema)) .await @@ -4252,6 +4256,7 @@ where total_rows: Some(total_steps as u64), status: TransferStatus::Running, error: None, + terminal: false, }); execute_on_pool(state, target_pool_key, &generate_postgres_enum_ddl(&enum_type, &request.target_schema)) .await @@ -4272,6 +4277,7 @@ where total_rows: Some(total_steps as u64), status: TransferStatus::Running, error: None, + terminal: false, }); execute_on_pool(state, target_pool_key, &generate_postgres_domain_ddl(&domain, &request.target_schema)) .await @@ -4373,6 +4379,7 @@ where total_rows: Some(total_steps as u64), status: TransferStatus::Running, error: None, + terminal: false, }); let rewritten_source = match object.object_type { @@ -4414,6 +4421,7 @@ where total_rows: Some(total_steps as u64), status: TransferStatus::Running, error: None, + terminal: false, }); execute_on_pool(state, target_pool_key, &statement) .await @@ -4435,6 +4443,7 @@ where total_rows: Some(total_steps as u64), status: TransferStatus::Running, error: None, + terminal: false, }); let full_table = qualified_table(&trigger.table_name, &request.target_schema, &DatabaseType::Postgres); let drop_sql = format!( @@ -4454,6 +4463,7 @@ where total_rows: Some(total_steps as u64), status: TransferStatus::Running, error: None, + terminal: false, }); let create_sql = rewrite_postgres_trigger_table_schema( &ensure_sql_statement_terminated(&trigger.source), @@ -4480,6 +4490,7 @@ where total_rows: Some(total_steps as u64), status: TransferStatus::Running, error: None, + terminal: false, }); execute_on_pool(state, target_pool_key, &statement) .await @@ -4500,6 +4511,7 @@ where total_rows: Some(total_steps as u64), status: TransferStatus::Running, error: None, + terminal: false, }); let ownership_owner = if matches!(request.ownership_policy, TransferOwnershipPolicy::ReassignMissing) && !ownership_existing_roles.contains(&statement.owner) @@ -4530,6 +4542,7 @@ where total_rows: Some(total_steps as u64), status: TransferStatus::Running, error: None, + terminal: false, }); execute_on_pool(state, target_pool_key, &statement) .await diff --git a/crates/dbx-web/src/routes/transfer.rs b/crates/dbx-web/src/routes/transfer.rs index 00d12b0b42..3f73fca471 100644 --- a/crates/dbx-web/src/routes/transfer.rs +++ b/crates/dbx-web/src/routes/transfer.rs @@ -130,6 +130,7 @@ pub async fn start_transfer( total_rows: None, status: TransferStatus::Cancelled, error: None, + terminal: true, }; if let Ok(json) = serde_json::to_string(&progress) { let _ = tx.send(json); @@ -148,6 +149,7 @@ pub async fn start_transfer( total_rows: None, status: TransferStatus::Error, error: Some(e), + terminal: true, }; if let Ok(json) = serde_json::to_string(&progress) { let _ = tx.send(json); @@ -170,6 +172,7 @@ pub async fn start_transfer( total_rows: None, status: TransferStatus::Cancelled, error: None, + terminal: true, }; if let Ok(json) = serde_json::to_string(&progress) { let _ = tx.send(json); @@ -212,6 +215,7 @@ pub async fn start_transfer( total_rows: last_total_rows.or(Some(rows)), status: TransferStatus::TableDone, error: None, + terminal: false, }; if let Ok(json) = serde_json::to_string(&progress) { let _ = tx.send(json); @@ -228,6 +232,7 @@ pub async fn start_transfer( total_rows: None, status: TransferStatus::Cancelled, error: None, + terminal: true, }; if let Ok(json) = serde_json::to_string(&progress) { let _ = tx.send(json); @@ -246,6 +251,7 @@ pub async fn start_transfer( total_rows: last_total_rows, status: TransferStatus::Error, error: Some(e), + terminal: false, }; if let Ok(json) = serde_json::to_string(&progress) { let _ = tx.send(json); @@ -282,6 +288,7 @@ pub async fn start_transfer( total_rows: None, status: TransferStatus::Cancelled, error: None, + terminal: true, }; if let Ok(json) = serde_json::to_string(&progress) { let _ = tx.send(json); @@ -301,6 +308,7 @@ pub async fn start_transfer( total_rows: None, status: TransferStatus::Error, error: Some(e), + terminal: false, }; if let Ok(json) = serde_json::to_string(&progress) { let _ = tx.send(json); @@ -327,6 +335,7 @@ pub async fn start_transfer( failed_tables.iter().take(5).cloned().collect::>().join(", ") )) }, + terminal: true, }; if let Ok(json) = serde_json::to_string(&done) { let _ = tx.send(json); diff --git a/packages/app-tests/transferProgress.test.ts b/packages/app-tests/transferProgress.test.ts new file mode 100644 index 0000000000..d5bbf99e90 --- /dev/null +++ b/packages/app-tests/transferProgress.test.ts @@ -0,0 +1,12 @@ +import { strict as assert } from "node:assert"; +import { test } from "vitest"; + +const { isTerminalTransferProgress } = await import("../../apps/desktop/src/lib/backend/transferProgress.ts"); + +test("treats only terminal transfer progress as the end of a transfer stream", () => { + assert.equal(isTerminalTransferProgress({ status: "error", terminal: false }), false); + assert.equal(isTerminalTransferProgress({ status: "error", terminal: true }), true); + assert.equal(isTerminalTransferProgress({ status: "done", terminal: false }), true); + assert.equal(isTerminalTransferProgress({ status: "cancelled", terminal: false }), true); + assert.equal(isTerminalTransferProgress({ status: "running", terminal: false }), false); +}); diff --git a/packages/app-tests/useExportTracker.test.ts b/packages/app-tests/useExportTracker.test.ts index 42d7a1374e..f9a0d69090 100644 --- a/packages/app-tests/useExportTracker.test.ts +++ b/packages/app-tests/useExportTracker.test.ts @@ -120,6 +120,7 @@ test("tracks data transfer progress and routes cancel to transfer API", async () totalRows: 500, status: "running", error: null, + terminal: false, }); assert.equal(task.kind, "data-transfer"); @@ -134,7 +135,7 @@ test("tracks data transfer progress and routes cancel to transfer API", async () assert.equal(apiMock.cancelTransfer.mock.calls[0][0], "transfer-1"); }); -test("keeps data transfer failed when final progress arrives after a table error", () => { +test("keeps data transfer active after a non-terminal table error", () => { const tracker = useExportTracker(); const task = tracker.addDataTransferTask("transfer-error-1", "source → target", 2); @@ -147,22 +148,67 @@ test("keeps data transfer failed when final progress arrives after a table error totalRows: null, status: "error", error: "permission denied", + terminal: false, }); + + assert.equal(task.status, "Running"); + assert.equal(task.errorMessage, "permission denied"); + + tracker.clearFinished(); + assert.equal( + tracker.tasks.value.some((item) => item.exportId === "transfer-error-1"), + true, + ); + tracker.updateDataTransferTask("transfer-error-1", { transferId: "transfer-error-1", + table: "orders", + tableIndex: 1, + totalTables: 2, + rowsTransferred: 10, + totalRows: 50, + status: "running", + error: null, + terminal: false, + }); + + assert.equal(task.status, "Running"); + assert.equal(task.errorMessage, "permission denied"); + assert.equal(task.currentTable, "orders"); + assert.equal(task.tableIndex, 1); + assert.equal(task.rowsExported, 10); +}); + +test("marks data transfer failed only on terminal error summary", () => { + const tracker = useExportTracker(); + const task = tracker.addDataTransferTask("transfer-terminal-error-1", "source → target", 2); + + tracker.updateDataTransferTask("transfer-terminal-error-1", { + transferId: "transfer-terminal-error-1", + table: "users", + tableIndex: 0, + totalTables: 2, + rowsTransferred: 0, + totalRows: null, + status: "error", + error: "permission denied", + terminal: false, + }); + tracker.updateDataTransferTask("transfer-terminal-error-1", { + transferId: "transfer-terminal-error-1", table: "", tableIndex: 2, totalTables: 2, rowsTransferred: 0, totalRows: null, - status: "done", - error: null, + status: "error", + error: "1 table(s) failed: users", + terminal: true, }); assert.equal(task.status, "Error"); - assert.equal(task.errorMessage, "permission denied"); + assert.equal(task.errorMessage, "1 table(s) failed: users"); assert.equal(task.tableIndex, 2); - assert.equal(task.rowsExported, 0); }); test("keeps data transfer row counts when terminal summary does not include rows", () => { @@ -178,6 +224,7 @@ test("keeps data transfer row counts when terminal summary does not include rows totalRows: 200, status: "tableDone", error: null, + terminal: false, }); tracker.updateDataTransferTask("transfer-rows-1", { transferId: "transfer-rows-1", @@ -188,6 +235,7 @@ test("keeps data transfer row counts when terminal summary does not include rows totalRows: null, status: "done", error: null, + terminal: true, }); assert.equal(task.status, "Done"); @@ -244,6 +292,7 @@ test("starts independent data transfer background tasks and routes progress by t totalRows: 100, status: "running", error: null, + terminal: false, }); callbacks.get("parallel-transfer-1")?.({ transferId: "parallel-transfer-1", @@ -254,6 +303,7 @@ test("starts independent data transfer background tasks and routes progress by t totalRows: 50, status: "running", error: null, + terminal: false, }); assert.equal(firstTask.currentTable, "users"); @@ -270,6 +320,7 @@ test("starts independent data transfer background tasks and routes progress by t totalRows: 50, status: "done", error: null, + terminal: true, }); resolvers.get("parallel-transfer-1")?.(); await new Promise((resolve) => setTimeout(resolve, 0)); @@ -319,5 +370,6 @@ test("blocks concurrent data transfers that write the same target table", () => totalRows: null, status: "done", error: null, + terminal: true, }); }); diff --git a/src-tauri/src/commands/transfer.rs b/src-tauri/src/commands/transfer.rs index f3055813b9..6a940139a4 100644 --- a/src-tauri/src/commands/transfer.rs +++ b/src-tauri/src/commands/transfer.rs @@ -79,6 +79,7 @@ pub async fn start_transfer( total_rows: None, status: TransferStatus::Cancelled, error: None, + terminal: true, }, ); dbx_core::transfer::clear_cancelled(&transfer_id).await; @@ -96,6 +97,7 @@ pub async fn start_transfer( total_rows: None, status: TransferStatus::Error, error: Some(e), + terminal: true, }, ); dbx_core::transfer::clear_cancelled(&transfer_id).await; @@ -118,6 +120,7 @@ pub async fn start_transfer( total_rows: None, status: TransferStatus::Cancelled, error: None, + terminal: true, }, ); dbx_core::transfer::clear_cancelled(&transfer_id).await; @@ -158,6 +161,7 @@ pub async fn start_transfer( total_rows: last_total_rows.or(Some(rows)), status: TransferStatus::TableDone, error: None, + terminal: false, }, ); } @@ -174,6 +178,7 @@ pub async fn start_transfer( total_rows: None, status: TransferStatus::Cancelled, error: None, + terminal: true, }, ); dbx_core::transfer::clear_cancelled(&transfer_id).await; @@ -191,6 +196,7 @@ pub async fn start_transfer( total_rows: last_total_rows, status: TransferStatus::Error, error: Some(e), + terminal: false, }, ); } @@ -222,6 +228,7 @@ pub async fn start_transfer( total_rows: None, status: TransferStatus::Cancelled, error: None, + terminal: true, }, ); dbx_core::transfer::clear_cancelled(&transfer_id).await; @@ -240,6 +247,7 @@ pub async fn start_transfer( total_rows: None, status: TransferStatus::Error, error: Some(e), + terminal: false, }, ); } @@ -265,6 +273,7 @@ pub async fn start_transfer( failed_tables.iter().take(5).cloned().collect::>().join(", ") )) }, + terminal: true, }, ); dbx_core::transfer::clear_cancelled(&transfer_id).await;