xiaoxue_admin
/
dbx
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(transfer): scope replay omission counts per subscriber

This commit is contained in:
t8y2 2026-07-28 12:43:36 +08:00
parent 8ac282cc37
commit 8d1472f7af
No known key found for this signature in database
2 changed files with 149 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
apps/desktop/src/composables/__tests__/useExportTracker.spec.ts vendored
View File

@ -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", () => {

116
crates/dbx-web/src/sse.rs
View File

@ -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<String>, broadcast::Receiver<String>) {
fn subscribe(&self) -> (Vec<String>, usize, broadcast::Receiver<String>) {
// 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::<Value>(&data) else {
return data;
let Ok(mut value) = serde_json::from_str::<Value>(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<TransferProgressChannel>,
) -> Sse<impl Stream<Item = Result<Event, std::convert::Infallible>>> {
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"#));
}
}