feat(transfer): add per-task operation timeout

apps/desktop/src/components/transfer/DataTransferDialog.vue

@@ -43,6 +43,7 @@ import { transferDatabaseTypeForConnection } from "@/lib/database/jdbcDialect";
 import { isDorisFamilyCatalogCapable } from "@/lib/database/databaseFeatureSupport";
 import { decodeTransferDatabaseOption, encodeTransferDatabaseOptions, formatTransferEndpointLabel, isSameTransferDatabase, isTransferDatabaseSelected, normalizeTransferCatalog } from "@/lib/database/dataTransferSelection";
 import { formatDatabaseLabel } from "@/lib/database/defaultDatabase";
+import { DEFAULT_QUERY_TIMEOUT_SECS, MAX_QUERY_TIMEOUT_SECS } from "@/lib/connection/timeoutLimits";
 import { databaseOptionsForConnection, fetchCatalogNamespaceOptions, fetchNamespaceOptionsForConnection, namespaceOptionsAreSchemas } from "@/composables/useDatabaseOptions";
 import { useExportTracker } from "@/composables/useExportTracker";
 import { openDataTransferTask } from "@/composables/useDialogSources";
@@ -85,6 +86,7 @@ const transferDialogStyle = {
 } as const;
 
 const store = useConnectionStore();
+const DEFAULT_TRANSFER_OPERATION_TIMEOUT_SECS = 600;
 
 const sqlConnections = computed(() => store.connections.filter((c) => supportsTransfer(transferDatabaseTypeForConnection(c))));
 const sourceConnections = computed(() => {
@@ -205,6 +207,8 @@ const targetTableStrategy = computed<TransferStrategy>({
 const targetTableNameCase = ref<TransferTableNameCase>("preserve");
 const quoteTargetColumnNames = ref(true);
 const batchSize = ref(1000);
+const operationTimeoutMode = ref<"inherit" | "custom">("inherit");
+const operationTimeoutSecs = ref(DEFAULT_TRANSFER_OPERATION_TIMEOUT_SECS);
 const showSqlPreviewConfirm = ref(false);
 // Per-source-table filter: table name -> bare WHERE predicate or full SELECT.
 const tableFilters = ref<Record<string, string>>({});
@@ -332,6 +336,26 @@ function databaseOptionLabel(connectionId: string, option: string): string {
 
 const sourceDatabaseName = computed(() => decodedDatabase(sourceConnectionId.value, sourceDatabase.value));
 const targetDatabaseName = computed(() => decodedDatabase(targetConnectionId.value, targetDatabase.value));
+const operationTimeoutValid = computed(() => operationTimeoutMode.value === "inherit" || (Number.isInteger(operationTimeoutSecs.value) && operationTimeoutSecs.value >= 0 && operationTimeoutSecs.value <= MAX_QUERY_TIMEOUT_SECS));
+
+function connectionQueryTimeoutSecs(connectionId: string): number {
+  const value = store.getConfig(connectionId)?.query_timeout_secs;
+  return typeof value === "number" && Number.isFinite(value) && value >= 0 ? Math.round(value) : DEFAULT_QUERY_TIMEOUT_SECS;
+}
+
+function formatOperationTimeout(timeoutSecs: number): string {
+  return timeoutSecs === 0 ? t("transfer.operationTimeoutUnlimited") : t("transfer.operationTimeoutSeconds", { seconds: timeoutSecs });
+}
+
+const operationTimeoutHint = computed(() => {
+  if (operationTimeoutMode.value === "custom") {
+    return operationTimeoutSecs.value === 0 ? t("transfer.operationTimeoutCustomUnlimitedHint") : t("transfer.operationTimeoutCustomHint", { seconds: operationTimeoutSecs.value });
+  }
+  return t("transfer.operationTimeoutInheritedHint", {
+    source: formatOperationTimeout(connectionQueryTimeoutSecs(sourceConnectionId.value)),
+    target: formatOperationTimeout(connectionQueryTimeoutSecs(targetConnectionId.value)),
+  });
+});
 
 const canStart = computed(() => {
   const effectiveSourceSchema = sourceSchema.value || sourceDatabaseName.value;
@@ -353,6 +377,7 @@ const canStart = computed(() => {
     (selectedTables.value.size > 0 || Object.values(selectedObjects.value).some((names) => names.size > 0)) &&
     (targetTableStrategy.value !== "rebuild" || !rebuildDisabledReason.value) &&
     (transferContent.value === "structureOnly" || targetTableStrategy.value !== "upsert" || upsertSupported.value) &&
+    operationTimeoutValid.value &&
     !sameSourceAndTarget
   );
 });
@@ -791,6 +816,8 @@ function resetState(cancelTaskLoad = true) {
   targetTableNameCase.value = "preserve";
   quoteTargetColumnNames.value = true;
   batchSize.value = 1000;
+  operationTimeoutMode.value = "inherit";
+  operationTimeoutSecs.value = DEFAULT_TRANSFER_OPERATION_TIMEOUT_SECS;
   tableFilters.value = {};
   filterDialogOpen.value = false;
   filterEditingTable.value = "";
@@ -912,6 +939,7 @@ async function requestStartTransfer() {
     quoteTargetColumnNames: quoteTargetColumnNames.value,
     ownershipPolicy: "preserve",
     batchSize: batchSize.value,
+    operationTimeoutSecs: operationTimeoutMode.value === "custom" ? operationTimeoutSecs.value : undefined,
     tableFilters: requestedTableFilters(),
     dropTargetConfirmed: false,
   };
@@ -994,6 +1022,7 @@ function currentConfig(): TransferTaskConfig {
     targetTableNameCase: targetTableNameCase.value,
     quoteTargetColumnNames: quoteTargetColumnNames.value,
     batchSize: batchSize.value,
+    operationTimeoutSecs: operationTimeoutMode.value === "custom" && operationTimeoutValid.value ? operationTimeoutSecs.value : undefined,
     tableFilters: requestedTableFilters(),
     dropTargetConfirmed: false,
   };
@@ -1012,7 +1041,7 @@ function configSnapshot(config: TransferTaskConfig) {
 const formHasContent = computed(() => !!sourceConnectionId.value || !!targetConnectionId.value);
 const isConfigDirty = computed(() => !!activeTaskId.value && configSnapshot(currentConfig()) !== savedConfigSnapshot.value);
 const needsDiscardConfirm = computed(() => isConfigDirty.value || (!activeTaskId.value && formHasContent.value));
-const canSaveConfig = computed(() => !!sourceConnectionId.value && isTransferDatabaseSelected(sourceDatabase.value) && !!targetConnectionId.value && isTransferDatabaseSelected(targetDatabase.value));
+const canSaveConfig = computed(() => !!sourceConnectionId.value && isTransferDatabaseSelected(sourceDatabase.value) && !!targetConnectionId.value && isTransferDatabaseSelected(targetDatabase.value) && operationTimeoutValid.value);
 
 // Invalidate before asynchronous callbacks can publish a preview for an old form,
 // including a task switch whose new values happen to equal the previous task.
@@ -1047,6 +1076,8 @@ async function loadTaskIntoForm(task: TransferTask) {
   targetTableNameCase.value = config.targetTableNameCase;
   quoteTargetColumnNames.value = config.quoteTargetColumnNames;
   batchSize.value = config.batchSize;
+  operationTimeoutMode.value = config.operationTimeoutSecs === undefined ? "inherit" : "custom";
+  operationTimeoutSecs.value = config.operationTimeoutSecs ?? DEFAULT_TRANSFER_OPERATION_TIMEOUT_SECS;
   pendingSelectedObjectsPrefill.value = Object.keys(config.objects).length > 0 ? JSON.parse(JSON.stringify(config.objects)) : null;
   pendingTableFiltersPrefill.value = config.tableFilters && Object.keys(config.tableFilters).length > 0 ? { ...config.tableFilters } : null;
 
@@ -1597,6 +1628,22 @@ async function saveConfigTask() {
                 <Label class="text-xs shrink-0">{{ t("transfer.batchSize") }}</Label>
                 <Input v-model.number="batchSize" type="number" min="100" max="10000" step="100" class="h-7 text-xs w-24" />
               </div>
+              <div class="space-y-1">
+                <div class="flex items-center gap-3">
+                  <Label class="text-xs shrink-0">{{ t("transfer.operationTimeout") }}</Label>
+                  <Select v-model="operationTimeoutMode">
+                    <SelectTrigger class="h-7 w-32 text-xs">
+                      <SelectValue />
+                    </SelectTrigger>
+                    <SelectContent>
+                      <SelectItem value="inherit">{{ t("transfer.operationTimeoutInherit") }}</SelectItem>
+                      <SelectItem value="custom">{{ t("transfer.operationTimeoutCustom") }}</SelectItem>
+                    </SelectContent>
+                  </Select>
+                  <Input v-if="operationTimeoutMode === 'custom'" v-model.number="operationTimeoutSecs" type="number" min="0" :max="MAX_QUERY_TIMEOUT_SECS" step="1" class="h-7 w-24 text-xs" />
+                </div>
+                <p class="text-xs leading-5 text-muted-foreground">{{ operationTimeoutHint }}</p>
+              </div>
             </div>
           </div>
         </div>

apps/desktop/src/i18n/locales/en.ts

@@ -7437,6 +7437,14 @@ export default {
     tableNameCaseLower: "Lowercase",
     tableNameCaseUpper: "Uppercase",
     batchSize: "Batch size",
+    operationTimeout: "Operation timeout (seconds)",
+    operationTimeoutInherit: "Inherit connections",
+    operationTimeoutCustom: "Custom",
+    operationTimeoutUnlimited: "Unlimited",
+    operationTimeoutSeconds: "{seconds} seconds",
+    operationTimeoutInheritedHint: "Source connection: {source}; target connection: {target}. The overall transfer has no total-duration limit.",
+    operationTimeoutCustomHint: "Source and target operations use a {seconds}-second timeout. Streaming reads treat it as an inactivity window; the overall transfer has no total-duration limit.",
+    operationTimeoutCustomUnlimitedHint: "Database operations in this transfer have no time limit. The task can still be cancelled manually.",
     tableFilterSection: "Per-table filters",
     tableFilterSectionHint: "Empty = full table",
     tableFilterSet: "Set filter",

apps/desktop/src/i18n/locales/zh-CN.ts

@@ -7419,6 +7419,14 @@ export default withEnglishFallback({
     tableNameCaseLower: "转为小写",
     tableNameCaseUpper: "转为大写",
     batchSize: "每批行数",
+    operationTimeout: "操作超时（秒）",
+    operationTimeoutInherit: "继承连接",
+    operationTimeoutCustom: "自定义",
+    operationTimeoutUnlimited: "不限制",
+    operationTimeoutSeconds: "{seconds} 秒",
+    operationTimeoutInheritedHint: "源连接：{source}；目标连接：{target}。整个传输任务不限制总时长。",
+    operationTimeoutCustomHint: "源端和目标端的单次操作使用 {seconds} 秒超时；流式读取按无进展时间计算，整个任务不限制总时长。",
+    operationTimeoutCustomUnlimitedHint: "此传输的数据库操作不限制时长；任务仍可手动取消。",
     tableFilterSection: "每表传输条件",
     tableFilterSectionHint: "不填则传全表",
     tableFilterSet: "设置条件",

apps/desktop/src/lib/backend/__tests__/dataTransferStart.spec.ts

@@ -43,7 +43,8 @@ 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 }));
+    const fetchMock = vi.fn().mockResolvedValue({ ok: true });
+    vi.stubGlobal("fetch", fetchMock);
     let source: FakeEventSource | undefined;
     class FakeEventSource {
       onmessage: ((event: { data: string }) => void) | null = null;
@@ -61,6 +62,7 @@ describe("web data-transfer submission", () => {
     const pending = startTransfer(request(), onProgress, onStarted);
     await vi.waitFor(() => expect(onStarted).toHaveBeenCalledTimes(1));
 
+    expect(JSON.parse(String(fetchMock.mock.calls[0]?.[1]?.body))).toEqual({ request: request() });
     expect(source?.url).toBe("/api/transfer/progress/transfer-1");
     source?.onmessage?.({ data: JSON.stringify(doneProgress()) });
     await expect(pending).resolves.toBeUndefined();
@@ -68,6 +70,25 @@ describe("web data-transfer submission", () => {
     expect(onProgress).toHaveBeenCalledWith(doneProgress());
   });
 
+  it("sends a transfer timeout outside the core request payload", async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ ok: true });
+    vi.stubGlobal("fetch", fetchMock);
+    class FakeEventSource {
+      onmessage: ((event: { data: string }) => void) | null = null;
+      onerror: (() => void) | null = null;
+      close = vi.fn();
+    }
+    vi.stubGlobal("EventSource", FakeEventSource);
+    const transferRequest = { ...request(), operationTimeoutSecs: 600 };
+
+    void startTransfer(transferRequest, vi.fn());
+    await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(1));
+
+    const body = JSON.parse(String(fetchMock.mock.calls[0]?.[1]?.body));
+    expect(body.operationTimeoutSecs).toBe(600);
+    expect(body.request.operationTimeoutSecs).toBeUndefined();
+  });
+
   it("suppresses acknowledgement when task creation fails", async () => {
     vi.stubGlobal(
       "fetch",

apps/desktop/src/lib/backend/http.ts

@@ -3065,11 +3065,12 @@ export async function saveGlobalSearchSettings(settings: GlobalSearchSettings):
 // ---------------------------------------------------------------------------
 
 export async function startTransfer(request: TransferRequest, onProgress: (progress: TransferProgress) => void, onStarted?: () => void): Promise<void> {
+  const { operationTimeoutSecs, ...backendRequest } = request;
   // 1. POST to start the transfer
   const res = await fetch(apiUrl("/api/transfer/start"), {
     method: "POST",
     headers: { "Content-Type": "application/json" },
-    body: JSON.stringify({ request }),
+    body: JSON.stringify({ request: backendRequest, operationTimeoutSecs }),
   });
   if (!res.ok) throw await backendResponseError(res);
   onStarted?.();

apps/desktop/src/lib/backend/tauri.ts

@@ -5799,6 +5799,8 @@ export interface TransferRequest {
   quoteTargetColumnNames: boolean;
   ownershipPolicy?: TransferOwnershipPolicy;
   batchSize: number;
+  /** Per-operation timeout for this transfer. Undefined inherits each connection; 0 disables the limit. */
+  operationTimeoutSecs?: number;
   /**
    * Optional per-source-table transfer filter.
    * Key = source table name; value = a bare `WHERE` predicate
@@ -5862,6 +5864,7 @@ export interface TransferProgress {
 }
 
 export async function startTransfer(request: TransferRequest, onProgress: (progress: TransferProgress) => void, onStarted?: () => void): Promise<void> {
+  const { operationTimeoutSecs, ...backendRequest } = request;
   return new Promise((resolve, reject) => {
     let unlisten: UnlistenFn | null = null;
     void (async () => {
@@ -5875,7 +5878,7 @@ export async function startTransfer(request: TransferRequest, onProgress: (progr
           }
         });
 
-        await invoke("start_transfer", { request });
+        await invoke("start_transfer", { request: backendRequest, operationTimeoutSecs });
         onStarted?.();
       } catch (e) {
         unlisten?.();

apps/desktop/src/stores/__tests__/transferTaskStore.spec.ts

@@ -115,6 +115,30 @@ describe("transferTaskStore", () => {
     expect(store.tasks[0]?.config.quoteTargetColumnNames).toBe(true);
   });
 
+  it("keeps legacy tasks on inherited timeouts and round-trips a custom timeout", async () => {
+    const store = useTransferTaskStore();
+    const inherited = await store.saveTask({ name: "inherited", config: makeConfig() });
+    const custom = await store.saveTask({ name: "custom", config: makeConfig({ operationTimeoutSecs: 600 }) });
+    const unlimited = await store.saveTask({ name: "unlimited", config: makeConfig({ operationTimeoutSecs: 0 }) });
+
+    expect(inherited.config.operationTimeoutSecs).toBeUndefined();
+    expect(custom.config.operationTimeoutSecs).toBe(600);
+    expect(unlimited.config.operationTimeoutSecs).toBe(0);
+  });
+
+  it("drops an invalid persisted transfer timeout instead of applying it", async () => {
+    vi.mocked(api.loadTransferTaskLibrary).mockResolvedValue({
+      version: 1,
+      folders: [],
+      tasks: [{ id: "invalid-timeout", name: "invalid timeout", config: makeConfig({ operationTimeoutSecs: 3601 }) }],
+    } as unknown as TransferTaskLibrary);
+    const store = useTransferTaskStore();
+
+    await store.initFromStorage();
+
+    expect(store.getTask("invalid-timeout")?.config.operationTimeoutSecs).toBeUndefined();
+  });
+
   it("loads a legacy rebuild task as append without restoring its previous confirmation", async () => {
     vi.mocked(api.loadTransferTaskLibrary).mockResolvedValue({
       version: 1,

apps/desktop/src/stores/transferTaskStore.ts

@@ -4,6 +4,7 @@ import { uuid } from "@/lib/common/utils";
 import * as api from "@/lib/backend/api";
 import type { TransferTask, TransferTaskConfig, TransferTaskFolder, TransferTaskLibrary } from "@/types/database";
 import { resolveTransferStrategy, transferStrategyOptions } from "@/components/transfer/transferStrategy";
+import { MAX_QUERY_TIMEOUT_SECS } from "@/lib/connection/timeoutLimits";
 
 export class TransferTaskNameConflictError extends Error {
   readonly code = "TRANSFER_TASK_NAME_CONFLICT";
@@ -70,6 +71,11 @@ function normalizeTransferTableFilters(raw: unknown): Record<string, string> | u
   return Object.keys(result).length > 0 ? result : undefined;
 }
 
+function normalizeTransferOperationTimeoutSecs(raw: unknown): number | undefined {
+  if (typeof raw !== "number" || !Number.isInteger(raw) || raw < 0 || raw > MAX_QUERY_TIMEOUT_SECS) return undefined;
+  return raw;
+}
+
 function normalizeTask(raw: unknown): TransferTask | null {
   if (!raw || typeof raw !== "object") return null;
   const candidate = raw as Partial<TransferTask>;
@@ -102,6 +108,7 @@ function normalizeTask(raw: unknown): TransferTask | null {
       targetTableNameCase: config.targetTableNameCase ?? "preserve",
       quoteTargetColumnNames: config.quoteTargetColumnNames ?? true,
       batchSize: typeof config.batchSize === "number" && config.batchSize > 0 ? config.batchSize : 1000,
+      operationTimeoutSecs: normalizeTransferOperationTimeoutSecs(config.operationTimeoutSecs),
       tableFilters: normalizeTransferTableFilters(config.tableFilters),
       dropTargetConfirmed: false,
     },
@@ -294,7 +301,12 @@ export const useTransferTaskStore = defineStore("transferTasks", () => {
     const folderId = Object.prototype.hasOwnProperty.call(input, "folderId") ? input.folderId || undefined : existing?.folderId;
     ensureTaskNameAvailable(trimmed, folderId, input.id);
     const timestamp = nowIso();
-    const config: TransferTaskConfig = { ...input.config, ...transferStrategyOptions(resolveTransferStrategy(input.config)), dropTargetConfirmed: false };
+    const config: TransferTaskConfig = {
+      ...input.config,
+      ...transferStrategyOptions(resolveTransferStrategy(input.config)),
+      operationTimeoutSecs: normalizeTransferOperationTimeoutSecs(input.config.operationTimeoutSecs),
+      dropTargetConfirmed: false,
+    };
     const task: TransferTask = existing
       ? { ...existing, folderId, name: trimmed, config, updatedAt: timestamp }
       : {

apps/desktop/src/types/database.ts

@@ -2524,6 +2524,8 @@ export interface TransferTaskConfig {
   targetTableNameCase: TransferTableNameCase;
   quoteTargetColumnNames: boolean;
   batchSize: number;
+  /** Per-operation timeout for this task. Undefined inherits each connection; 0 disables the limit. */
+  operationTimeoutSecs?: number;
   /** Optional per-source-table transfer filter (bare WHERE or a full SELECT). */
   tableFilters?: Record<string, string>;
   /** Legacy-compatible rebuild flag; true takes precedence over the saved DML mode. */

crates/dbx-core/src/data/transfer.rs

@@ -52,6 +52,36 @@ use crate::sql_dialect::{
 
 static CANCELLED: std::sync::LazyLock<RwLock<HashSet<String>>> =
     std::sync::LazyLock::new(|| RwLock::new(HashSet::new()));
+pub const MAX_TRANSFER_OPERATION_TIMEOUT_SECS: u64 = 3600;
+
+tokio::task_local! {
+    static TRANSFER_OPERATION_TIMEOUT_SECS: Option<u64>;
+}
+
+pub fn validate_transfer_operation_timeout(timeout_secs: Option<u64>) -> Result<(), String> {
+    if timeout_secs.is_some_and(|value| value > MAX_TRANSFER_OPERATION_TIMEOUT_SECS) {
+        return Err(format!(
+            "Transfer operation timeout must be between 0 and {MAX_TRANSFER_OPERATION_TIMEOUT_SECS} seconds"
+        ));
+    }
+    Ok(())
+}
+
+pub async fn with_transfer_operation_timeout<T>(
+    timeout_secs: Option<u64>,
+    future: impl std::future::Future<Output = T>,
+) -> T {
+    TRANSFER_OPERATION_TIMEOUT_SECS.scope(timeout_secs, future).await
+}
+
+fn current_transfer_operation_timeout_secs() -> Option<u64> {
+    TRANSFER_OPERATION_TIMEOUT_SECS.try_with(|timeout_secs| *timeout_secs).ok().flatten()
+}
+
+fn effective_transfer_operation_timeout_secs(connection_timeout_secs: u64) -> u64 {
+    current_transfer_operation_timeout_secs().unwrap_or(connection_timeout_secs)
+}
+
 static OCEANBASE_MYSQL_TABLE_OPTION_RE: std::sync::LazyLock<Regex> = std::sync::LazyLock::new(|| {
     Regex::new(r"(?i)\b(?:AUTO_INCREMENT_MODE|REPLICA_NUM|USE_BLOOM_FILTER|TABLET_SIZE|PCTFREE)\s*=")
         .expect("valid OceanBase MySQL table option regex")
@@ -7767,7 +7797,7 @@ async fn transfer_pool_context(
         config.map(|config| config.id.clone()),
         database_from_pool_key(pool_key).map(str::to_string),
         config.map(|config| config.db_type),
-        config.map(|config| config.effective_query_timeout_secs()),
+        config.map(|config| effective_transfer_operation_timeout_secs(config.effective_query_timeout_secs())),
     )
 }
 
@@ -10808,8 +10838,8 @@ async fn fetch_agent_transfer_batch(
     let query_timeout_secs = if cursor.started {
         0
     } else {
-        let configs = state.configs.read().await;
-        configs.get(&request.source_connection_id).map(|config| config.query_timeout_secs).unwrap_or(0)
+        let (_, _, _, timeout_secs) = transfer_pool_context(state, pool_key).await;
+        timeout_secs.unwrap_or(0)
     };
     let pool_handle = state.pool_handle(pool_key).await;
     let Some(PoolKind::Agent(client)) = pool_handle.as_ref() else {
@@ -12713,28 +12743,32 @@ where
     let request_target_catalog = request.target_catalog.clone();
     let (progress_tx, mut progress_rx) = tokio::sync::mpsc::channel(TRANSFER_PROGRESS_CHANNEL_CAPACITY);
     let (source_count_tx, mut source_count_rx) = tokio::sync::mpsc::channel(1);
+    let operation_timeout_secs = current_transfer_operation_timeout_secs();
 
     let mut task = tokio::spawn(async move {
         let mut task_pending_fk_alters = Vec::new();
-        let result = transfer_table_inner(
-            &state,
-            &request,
-            &table,
-            table_index,
-            &source_db_type,
-            &target_db_type,
-            &source_pool_key,
-            &target_pool_key,
-            &known_foreign_keys,
-            &mut task_pending_fk_alters,
-            preexisting_backup_names.as_ref(),
-            target_cleared,
-            move |progress| {
-                try_send_transfer_progress(&progress_tx, progress);
-            },
-            move |source_count| {
-                let _ = source_count_tx.try_send(source_count);
-            },
+        let result = with_transfer_operation_timeout(
+            operation_timeout_secs,
+            transfer_table_inner(
+                &state,
+                &request,
+                &table,
+                table_index,
+                &source_db_type,
+                &target_db_type,
+                &source_pool_key,
+                &target_pool_key,
+                &known_foreign_keys,
+                &mut task_pending_fk_alters,
+                preexisting_backup_names.as_ref(),
+                target_cleared,
+                move |progress| {
+                    try_send_transfer_progress(&progress_tx, progress);
+                },
+                move |source_count| {
+                    let _ = source_count_tx.try_send(source_count);
+                },
+            ),
         )
         .await;
         (result, task_pending_fk_alters)
@@ -13601,6 +13635,28 @@ for line in sys.stdin:
         assert!(!is_transfer_query_timeout("Connection timed out while loading tables"));
     }
 
+    #[test]
+    fn transfer_operation_timeout_validation_accepts_inherit_custom_and_unlimited() {
+        assert!(validate_transfer_operation_timeout(None).is_ok());
+        assert!(validate_transfer_operation_timeout(Some(0)).is_ok());
+        assert!(validate_transfer_operation_timeout(Some(MAX_TRANSFER_OPERATION_TIMEOUT_SECS)).is_ok());
+        assert!(validate_transfer_operation_timeout(Some(MAX_TRANSFER_OPERATION_TIMEOUT_SECS + 1)).is_err());
+    }
+
+    #[tokio::test]
+    async fn transfer_operation_timeout_override_is_task_scoped() {
+        assert_eq!(effective_transfer_operation_timeout_secs(60), 60);
+        assert_eq!(
+            with_transfer_operation_timeout(Some(600), async { effective_transfer_operation_timeout_secs(60) }).await,
+            600
+        );
+        assert_eq!(
+            with_transfer_operation_timeout(Some(0), async { effective_transfer_operation_timeout_secs(60) }).await,
+            0
+        );
+        assert_eq!(effective_transfer_operation_timeout_secs(60), 60);
+    }
+
     fn jdbc_transfer_config(connection_string: &str, driver_class: &str, profile: &str) -> ConnectionConfig {
         ConnectionConfig {
             oracle_oci_nls_lang: None,

crates/dbx-core/src/data/transfer_runner.rs

@@ -72,6 +72,7 @@ pub async fn run_transfer<F>(
     source_pool_key: String,
     target_pool_key: String,
     history: Option<TransferTaskJournal>,
+    operation_timeout_secs: Option<u64>,
     progress_callback: F,
 ) -> TransferProgress
 where
@@ -81,7 +82,7 @@ where
     let total_tables = request.tables.len();
     let app = ProgressEmitter::new(progress_callback);
 
-    async {
+    super::transfer::with_transfer_operation_timeout(operation_timeout_secs, async {
         // Sort tables by FK dependency so referenced tables are transferred first,
         // and keep the foreign key metadata fetched along the way — MySQL-family
         // targets reuse it per table below instead of re-querying it.
@@ -601,7 +602,7 @@ where
         )
         .await;
         super::transfer::clear_cancelled(&transfer_id).await;
-    }
+    })
     .await;
 
     super::transfer::clear_cancelled(&transfer_id).await;
@@ -672,6 +673,7 @@ mod tests {
             source_pool,
             target_pool,
             None,
+            None,
             move |progress| {
                 observed_for_callback.lock().unwrap_or_else(std::sync::PoisonError::into_inner).push(progress);
             },

crates/dbx-web/src/routes/transfer.rs

@@ -64,6 +64,7 @@ async fn finish_transfer_channel(state: &Arc<WebState>, transfer_id: &str, chann
 #[serde(rename_all = "camelCase")]
 pub struct StartTransferRequest {
     pub request: TransferRequest,
+    pub operation_timeout_secs: Option<u64>,
 }
 
 #[derive(Deserialize)]
@@ -82,8 +83,10 @@ pub async fn start_transfer(
     State(state): State<Arc<WebState>>,
     Json(body): Json<StartTransferRequest>,
 ) -> Result<Json<serde_json::Value>, AppError> {
+    let operation_timeout_secs = body.operation_timeout_secs;
     let req = body.request;
     transfer::validate_transfer_request(&req).map_err(AppError::from)?;
+    transfer::validate_transfer_operation_timeout(operation_timeout_secs).map_err(AppError::from)?;
 
     // Demo mode skips connection-backed validation: the simulated web build may not
     // carry the requested connections, and the spawned task reports failures as
@@ -188,7 +191,12 @@ pub async fn start_transfer(
             }
         };
 
-        if let Err(error) = transfer::ensure_transfer_source_types_supported(&app, &req, &source_pool_key).await {
+        if let Err(error) = transfer::with_transfer_operation_timeout(
+            operation_timeout_secs,
+            transfer::ensure_transfer_source_types_supported(&app, &req, &source_pool_key),
+        )
+        .await
+        {
             send_transfer_progress(&progress_channel, &terminal_transfer_error(&req, error));
             finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await;
             return;
@@ -204,6 +212,7 @@ pub async fn start_transfer(
             source_pool_key,
             target_pool_key,
             history,
+            operation_timeout_secs,
             move |progress| send_transfer_progress(&progress_channel_for_run, &progress),
         )
         .await;
@@ -436,12 +445,17 @@ mod tests {
 
         let req = transfer_request("src", "dst", &dir);
         let transfer_id = req.transfer_id.clone();
-        let response = start_transfer(State(state.clone()), Json(StartTransferRequest { request: req })).await.unwrap();
+        let response = start_transfer(
+            State(state.clone()),
+            Json(StartTransferRequest { request: req, operation_timeout_secs: None }),
+        )
+        .await
+        .unwrap();
         let _ = response.into_response();
 
         let duplicate = start_transfer(
             State(state.clone()),
-            Json(StartTransferRequest { request: transfer_request("src", "dst", &dir) }),
+            Json(StartTransferRequest { request: transfer_request("src", "dst", &dir), operation_timeout_secs: None }),
         )
         .await
         .unwrap_err();
@@ -511,7 +525,12 @@ mod tests {
         Arc::get_mut(&mut state).unwrap().demo_mode = true;
         let req = transfer_request("src", "dst", &dir);
         let transfer_id = req.transfer_id.clone();
-        let _ = start_transfer(State(state.clone()), Json(StartTransferRequest { request: req })).await.unwrap();
+        let _ = start_transfer(
+            State(state.clone()),
+            Json(StartTransferRequest { request: req, operation_timeout_secs: None }),
+        )
+        .await
+        .unwrap();
         let channel = {
             let channels = state.transfer_progress_channels.read().await;
             channels.get(&transfer_id).cloned().expect("transfer channel registered")

docs/content/docs/data-transfer.cn.mdx

@@ -83,9 +83,12 @@ description: 在不同连接、数据库、Schema 或数据库引擎之间批量
 | Upsert | 目标路径支持且能确定键时，按键更新或插入 |
 | 表名大小写 | 保持原样、转小写或转大写；转换后的目标名必须互不冲突 |
 | 批大小 | 控制分页读取和分批写入，默认 1000，可在界面中调整 |
+| 操作超时 | 默认分别继承源连接和目标连接的查询超时；也可为当前传输设置 `0`–`3600` 秒，`0` 表示不限制 |
 
 如果关闭“创建目标表”，目标表必须已经存在且字段可兼容。即使开启建表，跨引擎 DDL 也可能因目标数据库不支持某些类型、默认值、约束或对象而需要人工调整。
 
+操作超时不限制整个传输任务的总时长。支持增量结果流的驱动会把它作为“无进展等待时间”，每次收到数据都会重新计时；只能一次性返回结果的驱动则按单次数据库操作的总耗时计算。自定义值只作用于当前传输及其保存任务，不会修改连接或查询编辑器的超时设置。
+
 <Callout type="warn">**重建结构和数据**会替换现有目标表及其索引、触发器、约束、注释和对象权限为源结构。原表会先被重命名为备份表，传输成功后删除。任一步骤失败则保留备份表，可手工恢复。</Callout>
 
 ## 结构 SQL 预览

docs/content/docs/data-transfer.mdx

@@ -83,9 +83,12 @@ This entry point lists target drivers with implemented transactional writes. It
 | Upsert | Updates or inserts by key when the target path supports it and a usable key is available |
 | Target table name case | Preserve, lowercase, or uppercase; transformed target names must remain unique |
 | Batch size | Controls paged reads and batched writes; the UI default is 1000 |
+| Operation timeout | Inherits the source and target connection query timeouts by default; override this transfer with `0`–`3600` seconds, where `0` means unlimited |
 
 If table creation is disabled, compatible target tables must already exist. Even with table creation enabled, cross-engine DDL may require manual changes when the target does not support a source type, default, constraint, or object.
 
+The operation timeout does not cap the overall transfer duration. Drivers with incremental result streams treat it as an inactivity window that resets whenever data arrives; drivers that return a result all at once apply it to the total duration of each database operation. A custom value affects only this transfer and its saved task, not the connection or query-editor timeout settings.
+
 <Callout type="warn">**Rebuild structure and data** replaces existing target tables and their indexes, triggers, constraints, comments, and object permissions with source structure. The original table is renamed to a backup first, then dropped after a successful transfer. If any step fails, the backup table is retained and can be manually restored.</Callout>
 
 ## Structure SQL Preview

src-tauri/src/commands/transfer.rs

@@ -24,8 +24,10 @@ pub async fn start_transfer(
     app: AppHandle,
     state: State<'_, Arc<AppState>>,
     request: TransferRequest,
+    operation_timeout_secs: Option<u64>,
 ) -> Result<(), String> {
     let state = state.inner().clone();
+    dbx_core::transfer::validate_transfer_operation_timeout(operation_timeout_secs)?;
     // Reject transfer early if the target connection is read-only — writing to it is inherently required
     ensure_connection_writable(&state, &request.target_connection_id, "Transfer").await?;
 
@@ -73,7 +75,11 @@ pub async fn start_transfer(
     )
     .await?;
 
-    dbx_core::transfer::ensure_transfer_source_types_supported(&state, &request, &source_pool_key).await?;
+    dbx_core::transfer::with_transfer_operation_timeout(
+        operation_timeout_secs,
+        dbx_core::transfer::ensure_transfer_source_types_supported(&state, &request, &source_pool_key),
+    )
+    .await?;
 
     let history = match TransferTaskJournal::accept(&state.storage, &state, &request, TaskLifecycleOwner::Tauri).await {
         Ok(history) => history,
@@ -90,6 +96,7 @@ pub async fn start_transfer(
             source_pool_key,
             target_pool_key,
             history.clone(),
+            operation_timeout_secs,
             |progress| emit_progress(&app, progress),
         )
         .await;