From 8d1472f7af91617fc3cc13778a38309f7a49a005 Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Tue, 28 Jul 2026 12:43:36 +0800 Subject: [PATCH] fix(transfer): scope replay omission counts per subscriber --- .../__tests__/useExportTracker.spec.ts | 55 ++++++++- crates/dbx-web/src/sse.rs | 116 +++++++++++++++--- 2 files changed, 149 insertions(+), 22 deletions(-) diff --git a/apps/desktop/src/composables/__tests__/useExportTracker.spec.ts b/apps/desktop/src/composables/__tests__/useExportTracker.spec.ts index ce1383ca5..874f7972b 100644 --- a/apps/desktop/src/composables/__tests__/useExportTracker.spec.ts +++ b/apps/desktop/src/composables/__tests__/useExportTracker.spec.ts @@ -206,13 +206,58 @@ describe("data transfer failure details", () => { expect(task.transferFailuresOmitted).toBe(150); expect(task.transferFailures?.[0]).toEqual({ table: "table_0", error: "updated retained failure" }); + }); - tracker.updateDataTransferTask(task.exportId, { - ...transferProgress(task.exportId, "done"), - transferFailuresOmitted: 12, - }); + it("counts omitted failures once for an online subscription", () => { + const tracker = useExportTracker(); + const task = tracker.addDataTransferTask("online-failures", "many tables", 250); - expect(task.transferFailuresOmitted).toBe(162); + for (let index = 0; index < 250; index += 1) { + tracker.updateDataTransferTask(task.exportId, { + ...transferProgress(task.exportId, "error", false), + table: `table_${index}`, + error: `failure ${index}`, + }); + } + tracker.updateDataTransferTask(task.exportId, transferProgress(task.exportId, "done")); + + expect(task.transferFailures).toHaveLength(MAX_TRANSFER_FAILURE_DETAILS); + expect(task.transferFailuresOmitted).toBe(150); + }); + + it("combines pre-subscription and local omissions for a mid-transfer subscription", () => { + const tracker = useExportTracker(); + const task = tracker.addDataTransferTask("mid-transfer-failures", "many tables", 250); + + for (let index = 12; index < 250; index += 1) { + tracker.updateDataTransferTask(task.exportId, { + ...transferProgress(task.exportId, "error", false), + table: `table_${index}`, + error: `failure ${index}`, + ...(index === 249 ? { transferFailuresOmitted: 12 } : {}), + }); + } + tracker.updateDataTransferTask(task.exportId, transferProgress(task.exportId, "done")); + + expect(task.transferFailures).toHaveLength(MAX_TRANSFER_FAILURE_DETAILS); + expect(task.transferFailuresOmitted).toBe(150); + }); + + it("combines replay and local omissions for a delayed subscription", () => { + const tracker = useExportTracker(); + const task = tracker.addDataTransferTask("delayed-failures", "many tables", 250); + + for (let index = 12; index < 250; index += 1) { + tracker.updateDataTransferTask(task.exportId, { + ...transferProgress(task.exportId, "error", false), + table: `table_${index}`, + error: `failure ${index}`, + }); + } + tracker.updateDataTransferTask(task.exportId, { ...transferProgress(task.exportId, "done"), transferFailuresOmitted: 12 }); + + expect(task.transferFailures).toHaveLength(MAX_TRANSFER_FAILURE_DETAILS); + expect(task.transferFailuresOmitted).toBe(150); }); it("limits individual and total UTF-8 error detail bytes without splitting characters", () => { diff --git a/crates/dbx-web/src/sse.rs b/crates/dbx-web/src/sse.rs index d265153f0..20f8988a2 100644 --- a/crates/dbx-web/src/sse.rs +++ b/crates/dbx-web/src/sse.rs @@ -35,14 +35,13 @@ impl TransferProgressChannel { Self { tx, history: Mutex::new(TransferReplayHistory::default()) } } - pub fn send(&self, mut data: String, kind: TransferReplayEventKind) { + pub fn send(&self, data: String, kind: TransferReplayEventKind) { let mut history = self.history.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); match kind { TransferReplayEventKind::Progress => { history.latest = Some(data.clone()); } TransferReplayEventKind::Terminal => { - data = attach_omitted_failure_count(data, history.omitted_failures); history.latest = Some(data.clone()); } TransferReplayEventKind::Failure => { @@ -68,7 +67,7 @@ impl TransferProgressChannel { let _ = self.tx.send(data); } - fn subscribe(&self) -> (Vec, broadcast::Receiver) { + fn subscribe(&self) -> (Vec, usize, broadcast::Receiver) { // Subscribe while holding the history lock so an event is either in the // replay snapshot or in the live receiver, with no gap between them. let history = self.history.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); @@ -77,22 +76,26 @@ impl TransferProgressChannel { if let Some(latest) = &history.latest { replay.push(latest.clone()); } - (replay, rx) + (replay, history.omitted_failures, rx) } } -fn attach_omitted_failure_count(data: String, omitted_failures: usize) -> String { +fn attach_omitted_failure_count(data: &mut String, omitted_failures: usize) -> bool { if omitted_failures == 0 { - return data; + return false; } - let Ok(mut value) = serde_json::from_str::(&data) else { - return data; + let Ok(mut value) = serde_json::from_str::(data) else { + return false; }; let Some(object) = value.as_object_mut() else { - return data; + return false; }; object.insert("transferFailuresOmitted".to_string(), Value::from(omitted_failures as u64)); - serde_json::to_string(&value).unwrap_or(data) + let Ok(serialized) = serde_json::to_string(&value) else { + return false; + }; + *data = serialized; + true } pub fn sse_from_channel( @@ -147,14 +150,27 @@ pub fn sse_from_watch( pub fn sse_from_transfer_channel( channel: Arc, ) -> Sse>> { - let (replay, mut rx) = channel.subscribe(); + let (mut replay, mut pending_omitted_failures, mut rx) = channel.subscribe(); + if pending_omitted_failures > 0 { + for data in replay.iter_mut().rev() { + if attach_omitted_failure_count(data, pending_omitted_failures) { + pending_omitted_failures = 0; + break; + } + } + } let stream = async_stream::stream! { for data in replay { yield Ok(Event::default().data(data)); } loop { match rx.recv().await { - Ok(data) => yield Ok(Event::default().data(data)), + Ok(mut data) => { + if pending_omitted_failures > 0 && attach_omitted_failure_count(&mut data, pending_omitted_failures) { + pending_omitted_failures = 0; + } + yield Ok(Event::default().data(data)); + } Err(RecvError::Lagged(_)) => break, Err(RecvError::Closed) => break, } @@ -169,6 +185,16 @@ mod tests { use axum::body::to_bytes; use axum::response::IntoResponse; + fn failure_event(index: usize) -> String { + format!( + r#"{{"transferId":"transfer-1","table":"table-{index}","status":"error","error":"failure {index}","terminal":false}}"# + ) + } + + fn terminal_event() -> String { + r#"{"transferId":"transfer-1","status":"done","terminal":true}"#.to_string() + } + #[tokio::test] async fn delayed_transfer_subscription_replays_failure_and_terminal_events() { let channel = Arc::new(TransferProgressChannel::new()); @@ -188,17 +214,73 @@ mod tests { async fn delayed_transfer_subscription_reports_evicted_failure_count() { let channel = Arc::new(TransferProgressChannel::new()); for index in 0..=TRANSFER_REPLAY_MAX_FAILURES { - channel.send(format!("failure-{index}"), TransferReplayEventKind::Failure); + channel.send(failure_event(index), TransferReplayEventKind::Failure); } - channel.send( - r#"{"transferId":"transfer-1","status":"done","terminal":true}"#.to_string(), - TransferReplayEventKind::Terminal, - ); + channel.send(terminal_event(), TransferReplayEventKind::Terminal); let response = sse_from_transfer_channel(channel).into_response(); let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); let body = String::from_utf8(body.to_vec()).unwrap(); + assert_eq!(body.matches(r#""transferFailuresOmitted":1"#).count(), 1); + let terminal = body.lines().find(|line| line.contains(r#""status":"done""#)).unwrap(); + assert!(terminal.contains(r#""transferFailuresOmitted":1"#)); + } + + #[tokio::test] + async fn live_transfer_subscription_does_not_report_replay_omissions() { + let channel = Arc::new(TransferProgressChannel::new()); + let (replay, omitted_failures, mut rx) = channel.subscribe(); + assert!(replay.is_empty()); + assert_eq!(omitted_failures, 0); + + for index in 0..=TRANSFER_REPLAY_MAX_FAILURES { + let failure = failure_event(index); + channel.send(failure.clone(), TransferReplayEventKind::Failure); + assert_eq!(rx.recv().await.unwrap(), failure); + } + + let terminal = terminal_event(); + channel.send(terminal.clone(), TransferReplayEventKind::Terminal); + assert_eq!(rx.recv().await.unwrap(), terminal); + } + + #[tokio::test] + async fn mid_transfer_subscription_reports_only_pre_subscription_evictions() { + let channel = Arc::new(TransferProgressChannel::new()); + for index in 0..=TRANSFER_REPLAY_MAX_FAILURES { + channel.send(failure_event(index), TransferReplayEventKind::Failure); + } + + let response = sse_from_transfer_channel(channel.clone()).into_response(); + let terminal = terminal_event(); + channel.send(terminal.clone(), TransferReplayEventKind::Terminal); + drop(channel); + + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let body = String::from_utf8(body.to_vec()).unwrap(); + + assert_eq!(body.matches(r#""transferFailuresOmitted":1"#).count(), 1); + let terminal_line = body.lines().find(|line| line.contains(r#""status":"done""#)).unwrap(); + assert_eq!(terminal_line, format!("data: {terminal}")); + } + + #[tokio::test] + async fn omitted_failure_without_replay_is_reported_on_the_next_live_event() { + let channel = Arc::new(TransferProgressChannel::new()); + let oversized_failure = format!( + r#"{{"transferId":"transfer-1","table":"large","status":"error","error":"{}","terminal":false}}"#, + "x".repeat(TRANSFER_REPLAY_MAX_BYTES) + ); + channel.send(oversized_failure, TransferReplayEventKind::Failure); + + let response = sse_from_transfer_channel(channel.clone()).into_response(); + channel.send(terminal_event(), TransferReplayEventKind::Terminal); + drop(channel); + + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let body = String::from_utf8(body.to_vec()).unwrap(); + assert!(body.contains(r#""transferFailuresOmitted":1"#)); } }