github/dbx
4
0
Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-10-21 04:10:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(transfer): persist task run history

Browse Source
This commit is contained in:
jearhe
2026-10-09 09:44:41 +08:00
committed by GitHub
parent b755a5ca1e
commit 8b30a42b65
18 changed files with 2274 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/desktop/src/lib/backend/api.ts
+3
View File
@@ -931,6 +931,9 @@ export const saveHistory = forward("saveHistory");
export const loadHistory = forward("loadHistory");
export const searchHistory = forward("searchHistory");
export const loadHistoryConnectionOptions = forward("loadHistoryConnectionOptions");
export const loadTaskRuns = forward("loadTaskRuns");
export const loadTaskRun = forward("loadTaskRun");
export const loadTaskRunItems = forward("loadTaskRunItems");
export const loadRedisHistory = forward("loadRedisHistory");
export const clearHistory = forward("clearHistory");
export const clearHistoryBySource = forward("clearHistoryBySource");
apps/desktop/src/lib/backend/http.ts
+30
View File
@@ -151,6 +151,11 @@ import type {
HistorySearchRequest,
HistorySearchResult,
HistoryConnectionOption,
TaskRunDetail,
TaskRunItemsPage,
TaskRunItemsQuery,
TaskRunListQuery,
TaskRunPage,
SqlFileRequest,
SqlFilePreview,
SqlFileTable,
@@ -5300,6 +5305,31 @@ export async function loadHistoryConnectionOptions(): Promise<HistoryConnectionO
return get("/api/history/options");
}
export async function loadTaskRuns(query: TaskRunListQuery = {}): Promise<TaskRunPage> {
return get(
`/api/task-runs?${qs({
limit: query.limit,
cursorCreatedAt: query.cursor?.createdAt,
cursorRunId: query.cursor?.runId,
taskType: query.taskType,
status: query.status,
})}`,
);
}
export async function loadTaskRun(runId: string): Promise<TaskRunDetail | null> {
return get(`/api/task-runs/${encodeURIComponent(runId)}`);
}
export async function loadTaskRunItems(runId: string, query: TaskRunItemsQuery = {}): Promise<TaskRunItemsPage> {
return get(
`/api/task-runs/${encodeURIComponent(runId)}/items?${qs({
limit: query.limit,
afterItemIndex: query.afterItemIndex,
})}`,
);
}
export async function loadRedisHistory(limit = 100, offset = 0): Promise<HistoryEntry[]> {
return loadHistory(limit, offset, "redis_command");
}
apps/desktop/src/lib/backend/tauri.ts
+109
View File
@@ -5364,6 +5364,115 @@ export interface HistoryConnectionOption extends HistoryConnectionFilter {
databases: string[];
}
export type TaskType = "transfer";
export type TaskLifecycleOwner = "tauri" | "web";
export type TaskRunStatus = "running" | "succeeded" | "partial_failed" | "failed" | "cancelled";
export type TaskItemKind = "table" | "view" | "materialized_view" | "procedure" | "function" | "trigger" | "sequence" | "event" | "object";
export type TaskItemStatus = "pending" | "running" | "succeeded" | "skipped" | "failed" | "cancelled" | "not_started" | "incomplete";
export type TaskRowCountState = "not_applicable" | "known" | "unknown" | "incomplete";
export type TransferRunContent = "structure_and_data" | "structure_only" | "data_only";
export type TransferRunMode = "append" | "overwrite" | "upsert";
export type TransferRunObjectSelectionMode = "unspecified" | "explicit";
export interface TaskEndpointSnapshot {
connectionId: string;
databaseType: string;
database: string;
schema: string;
catalog?: string | null;
}
export interface TaskRun {
runId: string;
taskType: TaskType;
lifecycleOwner: TaskLifecycleOwner;
status: TaskRunStatus;
createdAt: string;
startedAt: string;
finishedAt?: string | null;
ownerInstanceId: string;
errorCode?: string | null;
safeErrorSummary?: string | null;
historyComplete: boolean;
source: TaskEndpointSnapshot;
target: TaskEndpointSnapshot;
}
export interface TransferRunDetails {
runId: string;
content: TransferRunContent;
mode: TransferRunMode;
batchSize: number;
createTable: boolean;
dropTargetBeforeCreate: boolean;
targetTableNameCase: "preserve" | "lower" | "upper";
quoteTargetColumnNames: boolean;
ownershipPolicy: "preserve" | "skip" | "reassign_missing";
filteredTableCount: number;
tableTotal: number;
objectSelectionMode: TransferRunObjectSelectionMode;
selectedObjectCount?: number | null;
}
export interface TaskRunDetail {
run: TaskRun;
transfer?: TransferRunDetails | null;
}
export interface TaskRunCursor {
createdAt: string;
runId: string;
}
export interface TaskRunListQuery {
limit?: number;
cursor?: TaskRunCursor | null;
taskType?: TaskType;
status?: TaskRunStatus;
}
export interface TaskRunPage {
items: TaskRun[];
nextCursor?: TaskRunCursor | null;
}
export interface TaskRunItem {
runId: string;
itemIndex: number;
itemKind: TaskItemKind;
sourceObject: string;
targetObject: string;
status: TaskItemStatus;
sourceRowCount?: number | null;
movedRowCount?: number | null;
targetRowCount?: number | null;
rowCountState: TaskRowCountState;
hasTableFilter: boolean;
safeErrorSummary?: string | null;
}
export interface TaskRunItemsQuery {
limit?: number;
afterItemIndex?: number;
}
export interface TaskRunItemsPage {
items: TaskRunItem[];
nextAfterItemIndex?: number | null;
}
export async function loadTaskRuns(query: TaskRunListQuery = {}): Promise<TaskRunPage> {
return invoke("load_task_runs", { query });
}
export async function loadTaskRun(runId: string): Promise<TaskRunDetail | null> {
return invoke("load_task_run", { runId });
}
export async function loadTaskRunItems(runId: string, query: TaskRunItemsQuery = {}): Promise<TaskRunItemsPage> {
return invoke("load_task_run_items", { runId, query });
}
export async function saveHistory(entry: HistoryEntry): Promise<void> {
return invoke("save_history", { entry });
}
crates/dbx-core/src/data/transfer.rs
+66 -9
View File
@@ -485,6 +485,13 @@ impl TransferRequest {
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TransferTableResult {
pub moved_rows: u64,
/// Source COUNT result when the executor obtained it. It is not a target row count.
pub source_row_count: Option<u64>,
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct TransferProgress {
@@ -10516,15 +10523,16 @@ async fn create_transfer_target_table(
Ok(())
}
/// Transfer a single table. Returns rows transferred.
/// `progress_callback` is invoked for progress updates.
/// Transfer a single table, returning moved rows and the source COUNT when available.
/// `progress_callback` is invoked for progress updates; source-count observation is separate
/// so the existing progress event payload remains unchanged.
///
/// `preexisting_backup_names` carries the output of [`rename_tables_to_backup`] and is
/// required whenever `drop_target_before_create` is set — this pass only checks whether the
/// table was renamed aside, and never renames or drops anything itself. Removing the backups
/// is [`drop_backup_tables`], after every table has succeeded.
#[allow(clippy::too_many_arguments)]
async fn transfer_table_inner<F>(
async fn transfer_table_inner<F, C>(
state: &Arc<AppState>,
request: &TransferRequest,
table: &str,
@@ -10537,14 +10545,17 @@ async fn transfer_table_inner<F>(
pending_fk_alters: &mut Vec<(String, String)>,
preexisting_backup_names: Option<&HashMap<String, String>>,
mut progress_callback: F,
) -> Result<u64, String>
mut source_count_callback: C,
) -> Result<TransferTableResult, String>
where
F: FnMut(TransferProgress),
C: FnMut(Option<u64>),
{
if *target_db_type == DatabaseType::Db2 {
db2::validate_request(request)?;
}
if is_mongodb_transfer_type(source_db_type) || is_mongodb_transfer_type(target_db_type) {
source_count_callback(None);
return transfer_mongodb_table(
state,
request,
@@ -10556,7 +10567,8 @@ where
target_pool_key,
progress_callback,
)
.await;
.await
.map(|moved_rows| TransferTableResult { moved_rows, source_row_count: None });
}
let table_filter = transfer_table_filter_for(request, table)?;
@@ -10762,6 +10774,7 @@ where
} else {
None
};
source_count_callback(total_rows);
log::info!("[transfer] {} total_rows={:?}", table, total_rows);
let server_side_complex_copy =
@@ -10847,7 +10860,7 @@ where
)
.await?;
}
return Ok(0);
return Ok(TransferTableResult { moved_rows: 0, source_row_count: None });
}
// A preexisting target also needs its columns read, even for a data-only
@@ -11032,7 +11045,7 @@ where
error: None,
terminal: false,
});
return Ok(copied);
return Ok(TransferTableResult { moved_rows: copied, source_row_count: total_rows });
}
// COPY fast path: PG-family append/overwrite transfers stream the whole
@@ -11394,7 +11407,7 @@ where
.await?;
}
Ok(total_transferred)
Ok(TransferTableResult { moved_rows: total_transferred, source_row_count: total_rows })
}
/// Free the constraint names the backups are still holding (MySQL family).
@@ -11705,10 +11718,49 @@ pub async fn transfer_table<F>(
known_foreign_keys: &HashMap<String, Vec<db::ForeignKeyInfo>>,
pending_fk_alters: &mut Vec<(String, String)>,
preexisting_backup_names: Option<&HashMap<String, String>>,
mut progress_callback: F,
progress_callback: F,
) -> Result<u64, String>
where
F: FnMut(TransferProgress),
{
transfer_table_with_result(
state,
request,
table,
table_index,
source_db_type,
target_db_type,
source_pool_key,
target_pool_key,
known_foreign_keys,
pending_fk_alters,
preexisting_backup_names,
progress_callback,
|_| {},
)
.await
.map(|result| result.moved_rows)
}
#[allow(clippy::too_many_arguments)]
pub async fn transfer_table_with_result<F, C>(
state: &Arc<AppState>,
request: &TransferRequest,
table: &str,
table_index: usize,
source_db_type: &DatabaseType,
target_db_type: &DatabaseType,
source_pool_key: &str,
target_pool_key: &str,
known_foreign_keys: &HashMap<String, Vec<db::ForeignKeyInfo>>,
pending_fk_alters: &mut Vec<(String, String)>,
preexisting_backup_names: Option<&HashMap<String, String>>,
mut progress_callback: F,
mut source_count_callback: C,
) -> Result<TransferTableResult, String>
where
F: FnMut(TransferProgress),
C: FnMut(Option<u64>),
{
let state = state.clone();
let request = request.clone();
@@ -11731,6 +11783,7 @@ where
let request_target_schema = request.target_schema.clone();
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 mut task = tokio::spawn(async move {
let mut task_pending_fk_alters = Vec::new();
@@ -11749,6 +11802,9 @@ where
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)
@@ -11759,6 +11815,7 @@ where
tokio::select! {
biased;
Some(progress) = progress_rx.recv() => progress_callback(progress),
Some(source_count) = source_count_rx.recv() => source_count_callback(source_count),
result = &mut task => {
let (result, task_pending_fk_alters) =
result.map_err(|error| format!("Transfer table task failed: {error}"))?;
crates/dbx-core/src/data/transfer/iris_tests.rs
+6 -1
View File
@@ -310,6 +310,7 @@ async fn iris_transfer_agent_preserves_errors_count_fallback_and_cancellation()
for failure in ["source", "target", "count", "cancel"] {
let (directory, state, request, source, target) = iris_transfer_fixture(None, 5, failure, true).await;
let mut totals = Vec::new();
let mut observed_source_count = None;
let result = Box::pin(transfer_table_inner(
&state,
&request,
@@ -328,12 +329,16 @@ async fn iris_transfer_agent_preserves_errors_count_fallback_and_cancellation()
CANCELLED.try_write().unwrap().insert(request.transfer_id.clone());
}
},
|source_count| observed_source_count = source_count,
))
.await;
clear_cancelled(&request.transfer_id).await;
state.shutdown(std::time::Duration::from_secs(1)).await;
if failure == "count" {
assert_eq!(result.unwrap(), 5);
let result = result.unwrap();
assert_eq!(result.moved_rows, 5);
assert_eq!(result.source_row_count, None);
assert_eq!(observed_source_count, None);
assert!(totals.iter().all(Option::is_none));
} else {
let error = result.unwrap_err();
crates/dbx-core/src/persistence/mod.rs
+1
View File
@@ -5,5 +5,6 @@ pub mod saved_sql;
pub mod secret_codec;
pub mod state_persistence;
pub mod storage;
pub mod task_history;
#[cfg(any(test, feature = "test-support"))]
pub mod test_storage;
crates/dbx-core/src/persistence/storage.rs
+663
View File
@@ -37,6 +37,12 @@ use crate::models::connection::{ConnectionConfig, DatabaseConnectionInfo, Databa
use crate::persistence::secret_codec::{
key_file_candidates, SecretCodec, SecretKeyPolicy, SecretKeyResolution, SecretKeySource,
};
use crate::persistence::task_history::{
RowCountState, TaskEndpointSnapshot, TaskHistoryStorageError, TaskItemKind, TaskItemStatus, TaskLifecycleOwner,
TaskRun, TaskRunCursor, TaskRunDetail, TaskRunItem, TaskRunItemsPage, TaskRunItemsQuery, TaskRunListQuery,
TaskRunPage, TaskRunStatus, TaskType, TransferObjectSelectionMode, TransferRunContent, TransferRunDetails,
TransferRunMode, TransferRunOwnershipPolicy, TransferRunTargetTableNameCase,
};
use crate::prompt_template::PromptTemplate;
use crate::saved_sql::{SavedSqlFile, SavedSqlFolder, SavedSqlLibrary};
@@ -83,6 +89,9 @@ const USER_DATA_TABLES: &[&str] = &[
"connections",
"connection_secrets",
"history",
"task_runs",
"transfer_run_details",
"task_run_items",
"ai_config",
"ai_provider_configs",
"ai_conversations",
@@ -1044,6 +1053,62 @@ const SCHEMA_STATEMENTS: &[&str] = &[
,mcp_response_json TEXT
,mcp_session_id TEXT
)",
"CREATE TABLE IF NOT EXISTS task_runs (
run_id TEXT PRIMARY KEY NOT NULL,
task_type TEXT NOT NULL,
lifecycle_owner TEXT NOT NULL,
status TEXT NOT NULL,
created_at TEXT NOT NULL,
started_at TEXT NOT NULL,
finished_at TEXT,
owner_instance_id TEXT NOT NULL,
error_code TEXT,
safe_error_summary TEXT,
history_complete INTEGER NOT NULL DEFAULT 0,
source_connection_id TEXT NOT NULL,
source_database_type TEXT NOT NULL,
source_database TEXT NOT NULL,
source_schema TEXT NOT NULL,
source_catalog TEXT,
target_connection_id TEXT NOT NULL,
target_database_type TEXT NOT NULL,
target_database TEXT NOT NULL,
target_schema TEXT NOT NULL,
target_catalog TEXT
)",
"CREATE INDEX IF NOT EXISTS idx_task_runs_created ON task_runs(created_at DESC, run_id DESC)",
"CREATE INDEX IF NOT EXISTS idx_task_runs_type_status_created ON task_runs(task_type, status, created_at DESC, run_id DESC)",
"CREATE TABLE IF NOT EXISTS transfer_run_details (
run_id TEXT PRIMARY KEY NOT NULL,
content TEXT NOT NULL,
mode TEXT NOT NULL,
batch_size INTEGER NOT NULL,
create_table INTEGER NOT NULL,
drop_target_before_create INTEGER NOT NULL,
target_table_name_case TEXT NOT NULL,
quote_target_column_names INTEGER NOT NULL,
ownership_policy TEXT NOT NULL,
filtered_table_count INTEGER NOT NULL,
table_total INTEGER NOT NULL,
object_selection_mode TEXT NOT NULL,
selected_object_count INTEGER
)",
"CREATE TABLE IF NOT EXISTS task_run_items (
run_id TEXT NOT NULL,
item_index INTEGER NOT NULL,
item_kind TEXT NOT NULL,
source_object TEXT NOT NULL,
target_object TEXT NOT NULL,
status TEXT NOT NULL,
source_row_count INTEGER,
moved_row_count INTEGER,
target_row_count INTEGER,
row_count_state TEXT NOT NULL,
has_table_filter INTEGER NOT NULL DEFAULT 0,
safe_error_summary TEXT,
PRIMARY KEY (run_id, item_index)
)",
"CREATE INDEX IF NOT EXISTS idx_task_run_items_run_status ON task_run_items(run_id, status, item_index)",
"CREATE TABLE IF NOT EXISTS ai_config (
id INTEGER PRIMARY KEY CHECK (id = 1),
config_json TEXT NOT NULL
@@ -3836,6 +3901,475 @@ impl Storage {
}
}
impl Storage {
pub async fn create_task_run(
&self,
run: &TaskRun,
details: &TransferRunDetails,
items: &[TaskRunItem],
) -> Result<(), TaskHistoryStorageError> {
let run = run.clone();
let details = details.clone();
let items = items.to_vec();
let inserted = self
.with_conn(move |conn| {
let tx = conn
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(|error| error.to_string())?;
let exists = tx
.query_row("SELECT 1 FROM task_runs WHERE run_id = ?1", [&run.run_id], |row| row.get::<_, i64>(0))
.optional()
.map_err(|error| error.to_string())?
.is_some();
if exists {
return Ok(false);
}
tx.execute(
"INSERT INTO task_runs (
run_id, task_type, lifecycle_owner, status, created_at, started_at, finished_at,
owner_instance_id, error_code, safe_error_summary, history_complete,
source_connection_id, source_database_type, source_database, source_schema, source_catalog,
target_connection_id, target_database_type, target_database, target_schema, target_catalog
) VALUES (
?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11,
?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21
)",
params![
run.run_id,
run.task_type.as_storage_str(),
run.lifecycle_owner.as_storage_str(),
run.status.as_storage_str(),
run.created_at,
run.started_at,
run.finished_at,
run.owner_instance_id,
run.error_code,
run.safe_error_summary,
run.history_complete,
run.source.connection_id,
run.source.database_type,
run.source.database,
run.source.schema,
run.source.catalog,
run.target.connection_id,
run.target.database_type,
run.target.database,
run.target.schema,
run.target.catalog,
],
)
.map_err(|error| error.to_string())?;
tx.execute(
"INSERT INTO transfer_run_details (
run_id, content, mode, batch_size, create_table, drop_target_before_create,
target_table_name_case, quote_target_column_names, ownership_policy,
filtered_table_count, table_total, object_selection_mode, selected_object_count
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
params![
details.run_id,
details.content.as_storage_str(),
details.mode.as_storage_str(),
details.batch_size,
details.create_table,
details.drop_target_before_create,
details.target_table_name_case.as_storage_str(),
details.quote_target_column_names,
details.ownership_policy.as_storage_str(),
details.filtered_table_count,
details.table_total,
details.object_selection_mode.as_storage_str(),
details.selected_object_count,
],
)
.map_err(|error| error.to_string())?;
for item in items {
insert_task_run_item(&tx, &item)?;
}
tx.commit().map_err(|error| error.to_string())?;
Ok(true)
})
.await
.map_err(|_| TaskHistoryStorageError::StorageUnavailable)?;
if inserted {
Ok(())
} else {
Err(TaskHistoryStorageError::RunIdConflict)
}
}
pub async fn save_task_run_item(&self, item: &TaskRunItem) -> Result<bool, TaskHistoryStorageError> {
let item = item.clone();
self.with_conn(move |conn| {
let tx = conn
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(|error| error.to_string())?;
let run_status = tx
.query_row("SELECT status FROM task_runs WHERE run_id = ?1", [&item.run_id], |row| {
row.get::<_, String>(0)
})
.optional()
.map_err(|error| error.to_string())?;
if run_status.as_deref() != Some("running") {
return Ok(false);
}
let changed = tx
.execute(
"INSERT INTO task_run_items (
run_id, item_index, item_kind, source_object, target_object, status,
source_row_count, moved_row_count, target_row_count, row_count_state,
has_table_filter, safe_error_summary
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)
ON CONFLICT(run_id, item_index) DO UPDATE SET
item_kind = excluded.item_kind,
source_object = excluded.source_object,
target_object = excluded.target_object,
status = excluded.status,
source_row_count = excluded.source_row_count,
moved_row_count = excluded.moved_row_count,
target_row_count = excluded.target_row_count,
row_count_state = excluded.row_count_state,
has_table_filter = excluded.has_table_filter,
safe_error_summary = excluded.safe_error_summary
WHERE task_run_items.status NOT IN ('succeeded', 'skipped', 'failed', 'cancelled', 'not_started', 'incomplete')",
params![
item.run_id,
item.item_index,
item.item_kind.as_storage_str(),
item.source_object,
item.target_object,
item.status.as_storage_str(),
item.source_row_count,
item.moved_row_count,
item.target_row_count,
item.row_count_state.as_storage_str(),
item.has_table_filter,
item.safe_error_summary,
],
)
.map_err(|error| error.to_string())?;
tx.commit().map_err(|error| error.to_string())?;
Ok(changed > 0)
})
.await
.map_err(|_| TaskHistoryStorageError::StorageUnavailable)
}
pub async fn finish_task_run(
&self,
run_id: &str,
status: TaskRunStatus,
error_code: Option<&str>,
safe_error_summary: Option<&str>,
history_complete: bool,
) -> Result<bool, TaskHistoryStorageError> {
let run_id = run_id.to_string();
let error_code = error_code.map(str::to_string);
let safe_error_summary = safe_error_summary.map(str::to_string);
self.with_conn(move |conn| {
let tx = conn
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(|error| error.to_string())?;
let current = tx
.query_row("SELECT status FROM task_runs WHERE run_id = ?1", [&run_id], |row| {
row.get::<_, String>(0)
})
.optional()
.map_err(|error| error.to_string())?;
let Some(current) = current else {
return Err("Task run does not exist".to_string());
};
if current != "running" {
return Ok(false);
}
if !status.is_terminal() {
return Err("A task run cannot be finished with a running status".to_string());
}
let finished_at = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
let changed = tx
.execute(
"UPDATE task_runs SET status = ?2, finished_at = ?3, error_code = ?4,
safe_error_summary = ?5, history_complete = ?6 WHERE run_id = ?1 AND status = 'running'",
params![run_id, status.as_storage_str(), finished_at, error_code, safe_error_summary, history_complete],
)
.map_err(|error| error.to_string())?;
if changed == 0 {
return Ok(false);
}
tx.execute(
"UPDATE task_run_items SET status = 'not_started'
WHERE run_id = ?1 AND status = 'pending'",
[&run_id],
)
.map_err(|error| error.to_string())?;
tx.execute(
"UPDATE task_run_items SET status = 'incomplete',
row_count_state = CASE WHEN row_count_state = 'not_applicable' THEN row_count_state ELSE 'incomplete' END,
safe_error_summary = COALESCE(safe_error_summary, ?2)
WHERE run_id = ?1 AND status = 'running'",
params![run_id, "Transfer item did not complete."],
)
.map_err(|error| error.to_string())?;
tx.commit().map_err(|error| error.to_string())?;
Ok(true)
})
.await
.map_err(|_| TaskHistoryStorageError::StorageUnavailable)
}
pub async fn get_task_run_detail(&self, run_id: &str) -> Result<Option<TaskRunDetail>, TaskHistoryStorageError> {
let run_id = run_id.to_string();
self.with_conn(move |conn| {
let run = conn
.query_row(
"SELECT run_id, task_type, lifecycle_owner, status, created_at, started_at, finished_at,
owner_instance_id, error_code, safe_error_summary, history_complete,
source_connection_id, source_database_type, source_database, source_schema, source_catalog,
target_connection_id, target_database_type, target_database, target_schema, target_catalog
FROM task_runs WHERE run_id = ?1",
[&run_id],
map_task_run_row,
)
.optional()
.map_err(|error| error.to_string())?;
let Some(run) = run else {
return Ok(None);
};
let transfer = conn
.query_row(
"SELECT run_id, content, mode, batch_size, create_table, drop_target_before_create,
target_table_name_case, quote_target_column_names, ownership_policy,
filtered_table_count, table_total, object_selection_mode, selected_object_count
FROM transfer_run_details WHERE run_id = ?1",
[&run_id],
map_transfer_run_details_row,
)
.optional()
.map_err(|error| error.to_string())?;
Ok(Some(TaskRunDetail { run, transfer }))
})
.await
.map_err(|_| TaskHistoryStorageError::StorageUnavailable)
}
pub async fn list_task_runs(&self, query: TaskRunListQuery) -> Result<TaskRunPage, TaskHistoryStorageError> {
self.with_conn(move |conn| {
let limit = query.limit.filter(|limit| *limit > 0).unwrap_or(50).clamp(1, 100);
let mut predicates = Vec::new();
let mut values = Vec::<Value>::new();
if let Some(task_type) = query.task_type {
predicates.push("task_type = ?".to_string());
values.push(Value::Text(task_type.as_storage_str().to_string()));
}
if let Some(status) = query.status {
predicates.push("status = ?".to_string());
values.push(Value::Text(status.as_storage_str().to_string()));
}
if let Some(cursor) = query.cursor {
predicates.push("(created_at < ? OR (created_at = ? AND run_id < ?))".to_string());
values.push(Value::Text(cursor.created_at.clone()));
values.push(Value::Text(cursor.created_at));
values.push(Value::Text(cursor.run_id));
}
let where_clause =
if predicates.is_empty() { String::new() } else { format!(" WHERE {}", predicates.join(" AND ")) };
let sql = format!(
"SELECT run_id, task_type, lifecycle_owner, status, created_at, started_at, finished_at,
owner_instance_id, error_code, safe_error_summary, history_complete,
source_connection_id, source_database_type, source_database, source_schema, source_catalog,
target_connection_id, target_database_type, target_database, target_schema, target_catalog
FROM task_runs{where_clause} ORDER BY created_at DESC, run_id DESC LIMIT ?"
);
values.push(Value::Integer((limit + 1) as i64));
let mut stmt = conn.prepare(&sql).map_err(|error| error.to_string())?;
let rows =
stmt.query_map(params_from_iter(values.iter()), map_task_run_row).map_err(|error| error.to_string())?;
let mut items = rows.collect::<Result<Vec<_>, _>>().map_err(|error| error.to_string())?;
let has_more = items.len() > limit;
items.truncate(limit);
let next_cursor = if has_more {
items.last().map(|run| TaskRunCursor { created_at: run.created_at.clone(), run_id: run.run_id.clone() })
} else {
None
};
Ok(TaskRunPage { items, next_cursor })
})
.await
.map_err(|_| TaskHistoryStorageError::StorageUnavailable)
}
pub async fn list_task_run_items(
&self,
run_id: &str,
query: TaskRunItemsQuery,
) -> Result<TaskRunItemsPage, TaskHistoryStorageError> {
let run_id = run_id.to_string();
self.with_conn(move |conn| {
let limit = query.limit.filter(|limit| *limit > 0).unwrap_or(100).clamp(1, 200);
let after = query.after_item_index.unwrap_or(-1);
let mut stmt = conn
.prepare(
"SELECT run_id, item_index, item_kind, source_object, target_object, status,
source_row_count, moved_row_count, target_row_count, row_count_state,
has_table_filter, safe_error_summary
FROM task_run_items WHERE run_id = ?1 AND item_index > ?2
ORDER BY item_index ASC LIMIT ?3",
)
.map_err(|error| error.to_string())?;
let rows = stmt
.query_map(params![run_id, after, (limit + 1) as i64], map_task_run_item_row)
.map_err(|error| error.to_string())?;
let mut items = rows.collect::<Result<Vec<_>, _>>().map_err(|error| error.to_string())?;
let has_more = items.len() > limit;
items.truncate(limit);
let next_after_item_index = if has_more { items.last().map(|item| item.item_index) } else { None };
Ok(TaskRunItemsPage { items, next_after_item_index })
})
.await
.map_err(|_| TaskHistoryStorageError::StorageUnavailable)
}
pub async fn delete_task_run(&self, run_id: &str) -> Result<bool, TaskHistoryStorageError> {
const RUNNING_SENTINEL: &str = "TASK_RUN_STILL_RUNNING";
let run_id = run_id.to_string();
let result = self
.with_conn(move |conn| {
let tx = conn
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(|error| error.to_string())?;
let current = tx
.query_row("SELECT status FROM task_runs WHERE run_id = ?1", [&run_id], |row| {
row.get::<_, String>(0)
})
.optional()
.map_err(|error| error.to_string())?;
let Some(current) = current else {
return Ok(false);
};
if TaskRunStatus::from_storage(&current).is_none_or(|status| !status.is_terminal()) {
return Err(RUNNING_SENTINEL.to_string());
}
tx.execute("DELETE FROM task_run_items WHERE run_id = ?1", [&run_id])
.map_err(|error| error.to_string())?;
tx.execute("DELETE FROM transfer_run_details WHERE run_id = ?1", [&run_id])
.map_err(|error| error.to_string())?;
tx.execute("DELETE FROM task_runs WHERE run_id = ?1", [&run_id]).map_err(|error| error.to_string())?;
tx.commit().map_err(|error| error.to_string())?;
Ok(true)
})
.await;
match result {
Ok(deleted) => Ok(deleted),
Err(error) if error == RUNNING_SENTINEL => Err(TaskHistoryStorageError::RunStillRunning),
Err(_) => Err(TaskHistoryStorageError::StorageUnavailable),
}
}
}
fn insert_task_run_item(tx: &rusqlite::Transaction<'_>, item: &TaskRunItem) -> Result<(), String> {
tx.execute(
"INSERT INTO task_run_items (
run_id, item_index, item_kind, source_object, target_object, status,
source_row_count, moved_row_count, target_row_count, row_count_state,
has_table_filter, safe_error_summary
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
params![
item.run_id,
item.item_index,
item.item_kind.as_storage_str(),
item.source_object,
item.target_object,
item.status.as_storage_str(),
item.source_row_count,
item.moved_row_count,
item.target_row_count,
item.row_count_state.as_storage_str(),
item.has_table_filter,
item.safe_error_summary,
],
)
.map(|_| ())
.map_err(|error| error.to_string())
}
fn map_task_run_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<TaskRun> {
let task_type: String = row.get(1)?;
let lifecycle_owner: String = row.get(2)?;
let status: String = row.get(3)?;
Ok(TaskRun {
run_id: row.get(0)?,
task_type: TaskType::from_storage(&task_type).ok_or(rusqlite::Error::InvalidQuery)?,
lifecycle_owner: TaskLifecycleOwner::from_storage(&lifecycle_owner).ok_or(rusqlite::Error::InvalidQuery)?,
status: TaskRunStatus::from_storage(&status).ok_or(rusqlite::Error::InvalidQuery)?,
created_at: row.get(4)?,
started_at: row.get(5)?,
finished_at: row.get(6)?,
owner_instance_id: row.get(7)?,
error_code: row.get(8)?,
safe_error_summary: row.get(9)?,
history_complete: row.get(10)?,
source: TaskEndpointSnapshot {
connection_id: row.get(11)?,
database_type: row.get(12)?,
database: row.get(13)?,
schema: row.get(14)?,
catalog: row.get(15)?,
},
target: TaskEndpointSnapshot {
connection_id: row.get(16)?,
database_type: row.get(17)?,
database: row.get(18)?,
schema: row.get(19)?,
catalog: row.get(20)?,
},
})
}
fn map_transfer_run_details_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<TransferRunDetails> {
let content: String = row.get(1)?;
let mode: String = row.get(2)?;
let table_name_case: String = row.get(6)?;
let ownership_policy: String = row.get(8)?;
let selection_mode: String = row.get(11)?;
Ok(TransferRunDetails {
run_id: row.get(0)?,
content: TransferRunContent::from_storage(&content).ok_or(rusqlite::Error::InvalidQuery)?,
mode: TransferRunMode::from_storage(&mode).ok_or(rusqlite::Error::InvalidQuery)?,
batch_size: row.get(3)?,
create_table: row.get(4)?,
drop_target_before_create: row.get(5)?,
target_table_name_case: TransferRunTargetTableNameCase::from_storage(&table_name_case)
.ok_or(rusqlite::Error::InvalidQuery)?,
quote_target_column_names: row.get(7)?,
ownership_policy: TransferRunOwnershipPolicy::from_storage(&ownership_policy)
.ok_or(rusqlite::Error::InvalidQuery)?,
filtered_table_count: row.get(9)?,
table_total: row.get(10)?,
object_selection_mode: TransferObjectSelectionMode::from_storage(&selection_mode)
.ok_or(rusqlite::Error::InvalidQuery)?,
selected_object_count: row.get(12)?,
})
}
fn map_task_run_item_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<TaskRunItem> {
let item_kind: String = row.get(2)?;
let status: String = row.get(5)?;
let row_count_state: String = row.get(9)?;
Ok(TaskRunItem {
run_id: row.get(0)?,
item_index: row.get(1)?,
item_kind: TaskItemKind::from_storage(&item_kind).ok_or(rusqlite::Error::InvalidQuery)?,
source_object: row.get(3)?,
target_object: row.get(4)?,
status: TaskItemStatus::from_storage(&status).ok_or(rusqlite::Error::InvalidQuery)?,
source_row_count: row.get(6)?,
moved_row_count: row.get(7)?,
target_row_count: row.get(8)?,
row_count_state: RowCountState::from_storage(&row_count_state).ok_or(rusqlite::Error::InvalidQuery)?,
has_table_filter: row.get(10)?,
safe_error_summary: row.get(11)?,
})
}
// AI Config
fn ai_provider_key(provider: &AiProvider) -> String {
@@ -14243,4 +14777,133 @@ mod tests {
.unwrap();
assert_eq!(storage.stored_connection_count().await.unwrap(), None);
}
#[tokio::test]
async fn task_run_storage_is_idempotent_paginated_and_deletes_only_terminal_runs() {
use crate::persistence::task_history::{
RowCountState, TaskEndpointSnapshot, TaskHistoryStorageError, TaskItemKind, TaskItemStatus,
TaskLifecycleOwner, TaskRun, TaskRunItem, TaskRunItemsQuery, TaskRunListQuery, TaskRunStatus, TaskType,
TransferObjectSelectionMode, TransferRunContent, TransferRunDetails, TransferRunMode,
TransferRunOwnershipPolicy, TransferRunTargetTableNameCase,
};
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("dbx.db");
let storage = crate::persistence::test_storage::open_unmigrated(&path).await.unwrap();
let run = TaskRun {
run_id: "history-storage-run".to_string(),
task_type: TaskType::Transfer,
lifecycle_owner: TaskLifecycleOwner::Web,
status: TaskRunStatus::Running,
created_at: "2025-01-01T00:00:00.000Z".to_string(),
started_at: "2025-01-01T00:00:00.000Z".to_string(),
finished_at: None,
owner_instance_id: "test-instance".to_string(),
error_code: None,
safe_error_summary: None,
history_complete: false,
source: TaskEndpointSnapshot {
connection_id: "source-id".to_string(),
database_type: "sqlite".to_string(),
database: "source-db".to_string(),
schema: "main".to_string(),
catalog: None,
},
target: TaskEndpointSnapshot {
connection_id: "target-id".to_string(),
database_type: "sqlite".to_string(),
database: "target-db".to_string(),
schema: "main".to_string(),
catalog: None,
},
};
let details = TransferRunDetails {
run_id: run.run_id.clone(),
content: TransferRunContent::DataOnly,
mode: TransferRunMode::Append,
batch_size: 500,
create_table: false,
drop_target_before_create: false,
target_table_name_case: TransferRunTargetTableNameCase::Preserve,
quote_target_column_names: true,
ownership_policy: TransferRunOwnershipPolicy::Preserve,
filtered_table_count: 0,
table_total: 1,
object_selection_mode: TransferObjectSelectionMode::Explicit,
selected_object_count: Some(0),
};
let item = TaskRunItem {
run_id: run.run_id.clone(),
item_index: 0,
item_kind: TaskItemKind::Table,
source_object: "orders".to_string(),
target_object: "orders".to_string(),
status: TaskItemStatus::Pending,
source_row_count: None,
moved_row_count: None,
target_row_count: None,
row_count_state: RowCountState::Unknown,
has_table_filter: false,
safe_error_summary: None,
};
storage.create_task_run(&run, &details, std::slice::from_ref(&item)).await.unwrap();
assert_eq!(
storage.create_task_run(&run, &details, std::slice::from_ref(&item)).await,
Err(TaskHistoryStorageError::RunIdConflict)
);
assert_eq!(storage.delete_task_run(&run.run_id).await, Err(TaskHistoryStorageError::RunStillRunning));
let completed_item = TaskRunItem {
status: TaskItemStatus::Succeeded,
source_row_count: Some(0),
moved_row_count: Some(0),
row_count_state: RowCountState::Known,
..item
};
assert!(storage.save_task_run_item(&completed_item).await.unwrap());
assert!(storage.finish_task_run(&run.run_id, TaskRunStatus::Succeeded, None, None, true).await.unwrap());
assert!(!storage
.finish_task_run(&run.run_id, TaskRunStatus::Failed, Some("late"), Some("late"), true)
.await
.unwrap());
let older_run = TaskRun {
run_id: "history-storage-run-older".to_string(),
created_at: "2024-12-31T00:00:00.000Z".to_string(),
started_at: "2024-12-31T00:00:00.000Z".to_string(),
..run.clone()
};
let older_details = TransferRunDetails { run_id: older_run.run_id.clone(), ..details.clone() };
storage.create_task_run(&older_run, &older_details, &[]).await.unwrap();
assert!(storage.finish_task_run(&older_run.run_id, TaskRunStatus::Succeeded, None, None, true).await.unwrap());
let page = storage.list_task_runs(TaskRunListQuery { limit: Some(1), ..Default::default() }).await.unwrap();
assert_eq!(page.items.len(), 1);
assert_eq!(page.items[0].run_id, run.run_id);
let cursor = page.next_cursor.clone().expect("older run must be available on the next page");
let next_page = storage
.list_task_runs(TaskRunListQuery { limit: Some(1), cursor: Some(cursor), ..Default::default() })
.await
.unwrap();
assert_eq!(next_page.items.len(), 1);
assert_eq!(next_page.items[0].run_id, older_run.run_id);
assert!(next_page.next_cursor.is_none());
let item_page = storage
.list_task_run_items(&run.run_id, TaskRunItemsQuery { limit: Some(1), ..Default::default() })
.await
.unwrap();
assert_eq!(item_page.items[0].source_row_count, Some(0));
assert_eq!(item_page.items[0].moved_row_count, Some(0));
assert_eq!(item_page.items[0].target_row_count, None);
drop(storage);
let reopened = crate::persistence::test_storage::open_unmigrated(&path).await.unwrap();
let detail = reopened.get_task_run_detail(&run.run_id).await.unwrap().unwrap();
assert_eq!(detail.run.status, TaskRunStatus::Succeeded);
assert!(detail.run.history_complete);
assert_eq!(detail.transfer.unwrap().selected_object_count, Some(0));
assert!(reopened.delete_task_run(&run.run_id).await.unwrap());
assert!(reopened.get_task_run_detail(&run.run_id).await.unwrap().is_none());
}
}
crates/dbx-core/src/persistence/task_history.rs
+1018
View File
File diff suppressed because it is too large Load Diff
crates/dbx-web/src/error.rs
+4
View File
@@ -26,6 +26,10 @@ impl AppError {
Self::with_status(msg.into(), StatusCode::BAD_REQUEST)
}
pub fn conflict(msg: impl Into<String>) -> Self {
Self::with_status(msg.into(), StatusCode::CONFLICT)
}
pub fn forbidden(msg: impl Into<String>) -> Self {
Self::with_status(msg.into(), StatusCode::FORBIDDEN)
}
crates/dbx-web/src/main.rs
+3
View File
@@ -1322,6 +1322,9 @@ async fn serve() -> Result<(), String> {
// Transfer
.route("/transfer/start", post(routes::transfer::start_transfer))
.route("/transfer/ownership-preview", post(routes::transfer::preview_transfer_ownership))
.route("/task-runs", get(routes::task_history::list_task_runs))
.route("/task-runs/{run_id}", get(routes::task_history::get_task_run))
.route("/task-runs/{run_id}/items", get(routes::task_history::list_task_run_items))
.route("/transfer/progress/{transferId}", get(routes::transfer::transfer_progress))
.route("/transfer/cancel", post(routes::transfer::cancel_transfer))
.route("/transfer/sort-tables-by-fk", post(routes::transfer::sort_tables_by_fk_dependency))
crates/dbx-web/src/routes/mod.rs
+1
View File
@@ -40,6 +40,7 @@ pub mod ssh_prompt;
pub mod tab_runtime_cache;
pub mod table_export;
pub mod table_import;
pub mod task_history;
pub mod text_export;
pub mod transfer;
pub mod tunnel_profiles;
crates/dbx-web/src/routes/task_history.rs
+75
View File
@@ -0,0 +1,75 @@
use std::sync::Arc;
use axum::extract::{Path, Query, State};
use axum::Json;
use dbx_core::persistence::task_history::{
TaskRunCursor, TaskRunDetail, TaskRunItemsPage, TaskRunItemsQuery, TaskRunListQuery, TaskRunPage, TaskRunStatus,
TaskType,
};
use serde::Deserialize;
use crate::error::AppError;
use crate::state::WebState;
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TaskRunListParams {
pub limit: Option<usize>,
pub cursor_created_at: Option<String>,
pub cursor_run_id: Option<String>,
pub task_type: Option<TaskType>,
pub status: Option<TaskRunStatus>,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TaskRunItemsParams {
pub limit: Option<usize>,
pub after_item_index: Option<i64>,
}
pub async fn list_task_runs(
State(state): State<Arc<WebState>>,
Query(params): Query<TaskRunListParams>,
) -> Result<Json<TaskRunPage>, AppError> {
let cursor = match (params.cursor_created_at, params.cursor_run_id) {
(Some(created_at), Some(run_id)) => Some(TaskRunCursor { created_at, run_id }),
(None, None) => None,
_ => return Err(AppError::bad_request("Both cursorCreatedAt and cursorRunId are required.")),
};
let query = TaskRunListQuery { limit: params.limit, cursor, task_type: params.task_type, status: params.status };
state.app.storage.list_task_runs(query).await.map(Json).map_err(|error| AppError::internal(error.code()))
}
pub async fn get_task_run(
State(state): State<Arc<WebState>>,
Path(run_id): Path<String>,
) -> Result<Json<TaskRunDetail>, AppError> {
state
.app
.storage
.get_task_run_detail(&run_id)
.await
.map_err(|error| AppError::internal(error.code()))?
.map(Json)
.ok_or_else(|| AppError::not_found("TASK_RUN_NOT_FOUND"))
}
pub async fn list_task_run_items(
State(state): State<Arc<WebState>>,
Path(run_id): Path<String>,
Query(params): Query<TaskRunItemsParams>,
) -> Result<Json<TaskRunItemsPage>, AppError> {
if state.app.storage.get_task_run_detail(&run_id).await.map_err(|error| AppError::internal(error.code()))?.is_none()
{
return Err(AppError::not_found("TASK_RUN_NOT_FOUND"));
}
let query = TaskRunItemsQuery { limit: params.limit, after_item_index: params.after_item_index };
state
.app
.storage
.list_task_run_items(&run_id, query)
.await
.map(Json)
.map_err(|error| AppError::internal(error.code()))
}
crates/dbx-web/src/routes/transfer.rs
+133 -4
View File
@@ -3,6 +3,9 @@ use std::sync::Arc;
use axum::extract::{Path, State};
use axum::response::sse::{Event, Sse};
use axum::Json;
use dbx_core::persistence::task_history::{
TaskHistoryStorageError, TaskItemStatus, TaskLifecycleOwner, TransferTaskJournal,
};
use dbx_core::transfer::{self, TransferRequest, TransferStatus};
use futures::stream::Stream;
use serde::Deserialize;
@@ -15,6 +18,9 @@ use crate::state::WebState;
const COMPLETED_TRANSFER_CHANNEL_TTL: Duration = Duration::from_secs(60);
fn send_transfer_progress(channel: &TransferProgressChannel, progress: &transfer::TransferProgress) {
if progress.terminal {
channel.remember_terminal_progress(progress.clone());
}
if let Ok(json) = serde_json::to_string(progress) {
let kind = if progress.terminal {
TransferReplayEventKind::Terminal
@@ -42,6 +48,7 @@ fn terminal_transfer_error(req: &TransferRequest, error: impl ToString) -> trans
}
async fn finish_transfer_channel(state: &Arc<WebState>, transfer_id: &str, channel: &Arc<TransferProgressChannel>) {
channel.finish_task_history().await;
transfer::clear_cancelled(transfer_id).await;
let state = state.clone();
let transfer_id = transfer_id.to_string();
@@ -106,10 +113,22 @@ pub async fn start_transfer(
}
let transfer_id = req.transfer_id.clone();
let history = if state.demo_mode {
None
} else {
match TransferTaskJournal::accept(&state.app.storage, &state.app, &req, TaskLifecycleOwner::Web).await {
Ok(history) => history,
Err(TaskHistoryStorageError::RunIdConflict) => {
return Err(AppError::conflict("TRANSFER_RUN_ID_CONFLICT"));
}
Err(error) => return Err(AppError::internal(error.code())),
}
};
// Keep bounded replay state so a web EventSource opened after this POST
// still receives early table failures and the terminal result.
let progress_channel = Arc::new(TransferProgressChannel::new());
progress_channel.set_task_history_journal(history.clone());
state.transfer_progress_channels.write().await.insert(transfer_id.clone(), progress_channel.clone());
let app = state.app.clone();
@@ -370,10 +389,19 @@ pub async fn start_transfer(
return;
}
let history_item_index =
history.as_ref().and_then(|journal| journal.item_index_for_table(table)).unwrap_or(i);
if let Some(journal) = history.as_ref() {
journal.start_table(history_item_index).await;
}
let progress_channel_clone = progress_channel.clone();
let mut last_rows_transferred = 0_u64;
let mut last_total_rows = None;
let result = transfer::transfer_table(
let mut source_row_count = None;
let mut moved_row_count = None;
let history_for_progress = history.as_ref();
let history_for_count = history.as_ref();
let result = transfer::transfer_table_with_result(
&app,
&req,
table,
@@ -388,20 +416,40 @@ pub async fn start_transfer(
|progress| {
last_rows_transferred = progress.rows_transferred;
last_total_rows = progress.total_rows;
moved_row_count = Some(progress.rows_transferred);
if let Some(journal) = history_for_progress {
journal.observe_table_progress(history_item_index, progress.rows_transferred);
}
send_transfer_progress(&progress_channel_clone, &progress);
},
|source_count| {
source_row_count = source_count;
if let Some(journal) = history_for_count {
journal.observe_source_count(history_item_index, source_count);
}
},
)
.await;
match result {
Ok(rows) => {
Ok(result) => {
if let Some(journal) = history.as_ref() {
journal
.finish_table(
history_item_index,
TaskItemStatus::Succeeded,
result.source_row_count,
Some(result.moved_rows),
)
.await;
}
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
table: table.clone(),
table_index: i,
total_tables: tables.len(),
rows_transferred: rows,
total_rows: last_total_rows.or(Some(rows)),
rows_transferred: result.moved_rows,
total_rows: last_total_rows.or(Some(result.moved_rows)),
status: TransferStatus::TableDone,
error: None,
terminal: false,
@@ -410,6 +458,16 @@ pub async fn start_transfer(
}
Err(e) => {
if e == "Cancelled" {
if let Some(journal) = history.as_ref() {
journal
.finish_table(
history_item_index,
TaskItemStatus::Cancelled,
source_row_count,
moved_row_count,
)
.await;
}
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
table: table.clone(),
@@ -425,6 +483,11 @@ pub async fn start_transfer(
finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await;
return;
}
if let Some(journal) = history.as_ref() {
journal
.finish_table(history_item_index, TaskItemStatus::Failed, source_row_count, moved_row_count)
.await;
}
failed_tables.push(table.clone());
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
@@ -476,9 +539,15 @@ pub async fn start_transfer(
.await
{
Ok(outcome) => {
if let Some(journal) = history.as_ref() {
journal.record_object_outcome(&outcome).await;
}
object_outcome = outcome;
}
Err(e) if e == "Cancelled" => {
if let Some(journal) = history.as_ref() {
journal.record_schema_objects_error(true).await;
}
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
table: "schema objects".to_string(),
@@ -495,6 +564,9 @@ pub async fn start_transfer(
return;
}
Err(e) => {
if let Some(journal) = history.as_ref() {
journal.record_schema_objects_error(false).await;
}
failed_tables.push("schema objects".to_string());
let progress = transfer::TransferProgress {
transfer_id: req.transfer_id.clone(),
@@ -673,6 +745,7 @@ mod tests {
default_connect_timeout_secs, default_idle_timeout_secs, default_keepalive_interval_secs,
default_query_timeout_secs, ConnectionConfig, DatabaseType,
};
use dbx_core::persistence::task_history::{TaskRunItemsQuery, TaskRunStatus, TransferObjectSelectionMode};
use dbx_core::transfer::{
TransferContent, TransferMode, TransferOwnershipPolicy, TransferRequest, TransferTableNameCase,
};
@@ -801,6 +874,14 @@ mod tests {
let response = start_transfer(State(state.clone()), Json(StartTransferRequest { request: req })).await.unwrap();
let _ = response.into_response();
let duplicate = start_transfer(
State(state.clone()),
Json(StartTransferRequest { request: transfer_request("src", "dst", &dir) }),
)
.await
.unwrap_err();
assert_eq!(duplicate.status, axum::http::StatusCode::CONFLICT);
// Both HTTP and Tauri call the same Core schema-object stage
// unconditionally. DataOnly must resolve there to a no-op without an
// object progress event or database-family fallback.
@@ -837,6 +918,54 @@ mod tests {
terminal_error.is_none(),
"DataOnly must complete without a schema-object error, got: {terminal_error:?}"
);
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
let task_run = loop {
let run = state.app.storage.get_task_run_detail(&transfer_id).await.unwrap();
if run.as_ref().is_some_and(|detail| detail.run.status.is_terminal()) {
break run.unwrap();
}
if std::time::Instant::now() > deadline {
panic!("task history did not reach a terminal state");
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
};
assert_eq!(task_run.run.lifecycle_owner, TaskLifecycleOwner::Web);
assert_eq!(task_run.run.status, TaskRunStatus::Succeeded);
assert!(task_run.run.history_complete);
let transfer_details = task_run.transfer.expect("accepted transfer details persisted");
assert_eq!(transfer_details.object_selection_mode, TransferObjectSelectionMode::Explicit);
assert_eq!(transfer_details.selected_object_count, Some(0));
let items = state.app.storage.list_task_run_items(&transfer_id, TaskRunItemsQuery::default()).await.unwrap();
assert!(items.items.is_empty());
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn demo_mode_transfer_does_not_write_task_history() {
let (mut state, dir) = test_web_state().await;
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 channel = {
let channels = state.transfer_progress_channels.read().await;
channels.get(&transfer_id).cloned().expect("transfer channel registered")
};
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
loop {
if channel
.latest()
.and_then(|data| serde_json::from_str::<serde_json::Value>(&data).ok())
.is_some_and(|progress| progress["terminal"].as_bool() == Some(true))
{
break;
}
if std::time::Instant::now() > deadline {
panic!("demo transfer did not reach a terminal event");
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
assert!(state.app.storage.get_task_run_detail(&transfer_id).await.unwrap().is_none());
let _ = std::fs::remove_dir_all(&dir);
}
}
crates/dbx-web/src/sse.rs
+22
View File
@@ -1,4 +1,6 @@
use axum::response::sse::{Event, KeepAlive, Sse};
use dbx_core::persistence::task_history::TransferTaskJournal;
use dbx_core::transfer::TransferProgress;
use futures::stream::Stream;
use serde_json::Value;
use std::collections::VecDeque;
@@ -22,6 +24,8 @@ struct TransferReplayHistory {
failure_bytes: usize,
latest: Option<String>,
omitted_failures: usize,
task_history_journal: Option<TransferTaskJournal>,
terminal_progress: Option<TransferProgress>,
}
pub struct TransferProgressChannel {
@@ -35,6 +39,24 @@ impl TransferProgressChannel {
Self { tx, history: Mutex::new(TransferReplayHistory::default()) }
}
pub fn set_task_history_journal(&self, journal: Option<TransferTaskJournal>) {
self.history.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).task_history_journal = journal;
}
pub fn remember_terminal_progress(&self, progress: TransferProgress) {
self.history.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).terminal_progress = Some(progress);
}
pub async fn finish_task_history(&self) {
let (journal, progress) = {
let history = self.history.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
(history.task_history_journal.clone(), history.terminal_progress.clone())
};
if let (Some(journal), Some(progress)) = (journal, progress) {
journal.finish(&progress).await;
}
}
pub fn send(&self, data: String, kind: TransferReplayEventKind) {
let mut history = self.history.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
match kind {
src-tauri/src/commands/mod.rs
+1
View File
@@ -64,6 +64,7 @@ pub mod system_fonts;
pub mod tab_runtime_cache;
pub mod table_export;
pub mod table_import;
pub mod task_history;
pub mod text_export;
pub mod transfer;
pub mod tunnel_profiles;
src-tauri/src/commands/task_history.rs
+27
View File
@@ -0,0 +1,27 @@
use std::sync::Arc;
use tauri::State;
use super::connection::AppState;
use dbx_core::persistence::task_history::{
TaskRunDetail, TaskRunItemsPage, TaskRunItemsQuery, TaskRunListQuery, TaskRunPage,
};
#[tauri::command]
pub async fn load_task_runs(state: State<'_, Arc<AppState>>, query: TaskRunListQuery) -> Result<TaskRunPage, String> {
state.storage.list_task_runs(query).await.map_err(|error| error.code().to_string())
}
#[tauri::command]
pub async fn load_task_run(state: State<'_, Arc<AppState>>, run_id: String) -> Result<Option<TaskRunDetail>, String> {
state.storage.get_task_run_detail(&run_id).await.map_err(|error| error.code().to_string())
}
#[tauri::command]
pub async fn load_task_run_items(
state: State<'_, Arc<AppState>>,
run_id: String,
query: TaskRunItemsQuery,
) -> Result<TaskRunItemsPage, String> {
state.storage.list_task_run_items(&run_id, query).await.map_err(|error| error.code().to_string())
}
src-tauri/src/commands/transfer.rs
+109 -25
View File
@@ -5,6 +5,9 @@ use crate::commands::connection::{ensure_connection_writable, AppState};
// Re-export types and functions used by other modules
use dbx_core::models::connection::DatabaseType;
use dbx_core::persistence::task_history::{
TaskHistoryStorageError, TaskItemStatus, TaskLifecycleOwner, TransferTaskJournal,
};
pub use dbx_core::transfer::{
get_db_type, TransferOwnershipPreview, TransferProgress, TransferRequest, TransferStatus,
};
@@ -13,6 +16,13 @@ fn emit_progress(app: &AppHandle, progress: TransferProgress) {
let _ = app.emit("transfer-progress", progress);
}
async fn emit_terminal_progress(app: &AppHandle, history: Option<&TransferTaskJournal>, progress: TransferProgress) {
if let Some(history) = history {
history.finish(&progress).await;
}
emit_progress(app, progress);
}
#[tauri::command]
pub async fn start_transfer(
app: AppHandle,
@@ -70,6 +80,12 @@ pub async fn start_transfer(
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,
Err(TaskHistoryStorageError::RunIdConflict) => return Err("TRANSFER_RUN_ID_CONFLICT".to_string()),
Err(error) => return Err(error.code().to_string()),
};
tokio::spawn(async move {
// Sort tables by FK dependency so referenced tables are transferred first,
// and keep the foreign key metadata fetched along the way — MySQL-family
@@ -157,8 +173,9 @@ pub async fn start_transfer(
{
Ok(names) => Some(names),
Err(e) if e == "Cancelled" => {
emit_progress(
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "rename pre-pass".to_string(),
@@ -170,13 +187,15 @@ pub async fn start_transfer(
error: None,
terminal: true,
},
);
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
return;
}
Err(e) => {
emit_progress(
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "rename pre-pass".to_string(),
@@ -188,7 +207,8 @@ pub async fn start_transfer(
error: Some(e),
terminal: true,
},
);
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
return;
}
@@ -215,8 +235,9 @@ pub async fn start_transfer(
{
Ok(()) => {}
Err(e) if e == "Cancelled" => {
emit_progress(
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "schema dependencies".to_string(),
@@ -228,13 +249,15 @@ pub async fn start_transfer(
error: None,
terminal: true,
},
);
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
return;
}
Err(e) => {
emit_progress(
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "schema dependencies".to_string(),
@@ -246,7 +269,8 @@ pub async fn start_transfer(
error: Some(e),
terminal: true,
},
);
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
return;
}
@@ -254,8 +278,9 @@ pub async fn start_transfer(
}
for (i, table) in sorted_tables.iter().enumerate() {
if dbx_core::transfer::is_cancelled(&transfer_id).await {
emit_progress(
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: table.clone(),
@@ -267,14 +292,23 @@ pub async fn start_transfer(
error: None,
terminal: true,
},
);
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
return;
}
log::info!("[transfer] table {}/{}: {}", i + 1, total_tables, table);
match dbx_core::transfer::transfer_table(
let history_item_index =
history.as_ref().and_then(|journal| journal.item_index_for_table(table)).unwrap_or(i);
if let Some(journal) = history.as_ref() {
journal.start_table(history_item_index).await;
}
let mut source_row_count = None;
let mut moved_row_count = None;
let history_for_progress = history.as_ref();
let history_for_count = history.as_ref();
let result = dbx_core::transfer::transfer_table_with_result(
&state,
&request,
table,
@@ -289,12 +323,32 @@ pub async fn start_transfer(
|progress| {
last_rows_transferred = progress.rows_transferred;
last_total_rows = progress.total_rows;
moved_row_count = Some(progress.rows_transferred);
if let Some(journal) = history_for_progress {
journal.observe_table_progress(history_item_index, progress.rows_transferred);
}
emit_progress(&app, progress);
},
|source_count| {
source_row_count = source_count;
if let Some(journal) = history_for_count {
journal.observe_source_count(history_item_index, source_count);
}
},
)
.await
{
Ok(rows) => {
.await;
match result {
Ok(result) => {
if let Some(journal) = history.as_ref() {
journal
.finish_table(
history_item_index,
TaskItemStatus::Succeeded,
result.source_row_count,
Some(result.moved_rows),
)
.await;
}
emit_progress(
&app,
TransferProgress {
@@ -302,8 +356,8 @@ pub async fn start_transfer(
table: table.clone(),
table_index: i,
total_tables,
rows_transferred: rows,
total_rows: last_total_rows.or(Some(rows)),
rows_transferred: result.moved_rows,
total_rows: last_total_rows.or(Some(result.moved_rows)),
status: TransferStatus::TableDone,
error: None,
terminal: false,
@@ -312,23 +366,40 @@ pub async fn start_transfer(
}
Err(e) => {
if e == "Cancelled" {
emit_progress(
if let Some(journal) = history.as_ref() {
journal
.finish_table(
history_item_index,
TaskItemStatus::Cancelled,
source_row_count,
moved_row_count,
)
.await;
}
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: table.clone(),
table_index: i,
total_tables,
rows_transferred: last_rows_transferred,
total_rows: last_total_rows,
rows_transferred: 0,
total_rows: None,
status: TransferStatus::Cancelled,
error: None,
terminal: true,
},
);
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
return;
}
if let Some(journal) = history.as_ref() {
journal
.finish_table(history_item_index, TaskItemStatus::Failed, source_row_count, moved_row_count)
.await;
}
failed_tables.push(table.clone());
emit_progress(
&app,
@@ -385,11 +456,18 @@ pub async fn start_transfer(
.await
{
Ok(outcome) => {
if let Some(journal) = history.as_ref() {
journal.record_object_outcome(&outcome).await;
}
object_outcome = outcome;
}
Err(e) if e == "Cancelled" => {
emit_progress(
if let Some(journal) = history.as_ref() {
journal.record_schema_objects_error(true).await;
}
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: "schema objects".to_string(),
@@ -401,11 +479,15 @@ pub async fn start_transfer(
error: None,
terminal: true,
},
);
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
return;
}
Err(e) => {
if let Some(journal) = history.as_ref() {
journal.record_schema_objects_error(false).await;
}
failed_tables.push("schema objects".to_string());
emit_progress(
&app,
@@ -461,8 +543,9 @@ pub async fn start_transfer(
String::new()
};
emit_progress(
emit_terminal_progress(
&app,
history.as_ref(),
TransferProgress {
transfer_id: transfer_id.clone(),
table: String::new(),
@@ -487,7 +570,8 @@ pub async fn start_transfer(
},
terminal: true,
},
);
)
.await;
dbx_core::transfer::clear_cancelled(&transfer_id).await;
});
src-tauri/src/lib.rs
+3
View File
@@ -2583,6 +2583,9 @@ pub fn run() {
commands::history::clear_history_by_source,
commands::history::cleanup_mcp_history_retention,
commands::history::delete_history_entry,
commands::task_history::load_task_runs,
commands::task_history::load_task_run,
commands::task_history::load_task_run_items,
commands::mcp::check_mcp_server_status,
commands::mcp::install_mcp_server,
commands::mcp::install_native_mcp_server,