diff --git a/crates/ely_app/src/shell/bookmark_files.rs b/crates/ely_app/src/shell/bookmark_files.rs index de7e605..f7b85eb 100644 --- a/crates/ely_app/src/shell/bookmark_files.rs +++ b/crates/ely_app/src/shell/bookmark_files.rs @@ -184,7 +184,10 @@ impl ElyShell { }; match import_result { - Ok(message) => self.set_bookmark_file_notice(message, cx), + Ok(message) => { + self.set_bookmark_file_notice(message, cx); + self.schedule_cloud_sync_upload(cx); + } Err(error) => self.set_bookmark_file_error(error, cx), } } diff --git a/crates/ely_app/src/shell/bookmarks.rs b/crates/ely_app/src/shell/bookmarks.rs index 22fe0a5..0820d4c 100644 --- a/crates/ely_app/src/shell/bookmarks.rs +++ b/crates/ely_app/src/shell/bookmarks.rs @@ -91,9 +91,11 @@ impl ElyShell { ShellState::StartupError(message) => Err(message.clone()), }; + let updated = result.is_ok(); self.bookmark_edit_error = result.err(); - if self.bookmark_edit_error.is_none() { + if updated { self.pending_bookmark_edit = None; + self.schedule_cloud_sync_upload(cx); } cx.notify(); } diff --git a/crates/ely_app/src/shell/mod.rs b/crates/ely_app/src/shell/mod.rs index a2814c9..10a5033 100644 --- a/crates/ely_app/src/shell/mod.rs +++ b/crates/ely_app/src/shell/mod.rs @@ -105,6 +105,10 @@ pub struct ElyShell { /// `tick_external_web_surfaces`. sync_inbox_rx: std::sync::mpsc::Receiver, pub(crate) sync_inbox_tx: std::sync::mpsc::Sender, + sync_upload_scheduled: bool, + sync_upload_in_flight: bool, + sync_upload_pending: bool, + sync_upload_pending_logical_clock_floor: Option, pub(crate) auth_email_input: Entity, pub(crate) auth_otp_input: Entity, pub(crate) auth_flow_phase: auth::AuthFlowPhase, @@ -178,6 +182,7 @@ impl ElyShell { if sync_address { shell.sync_address_input(window, cx); + shell.schedule_cloud_sync_upload(cx); } cx.notify(); @@ -231,6 +236,10 @@ impl ElyShell { web_surfaces: WebSurfaceStore::new(), sync_inbox_rx, sync_inbox_tx, + sync_upload_scheduled: false, + sync_upload_in_flight: false, + sync_upload_pending: false, + sync_upload_pending_logical_clock_floor: None, auth_email_input, auth_otp_input, auth_flow_phase: auth::AuthFlowPhase::Idle, @@ -245,84 +254,6 @@ impl ElyShell { shell } - /// Drain any sync upload outcomes the off-thread worker pushed - /// since the previous tick and stamp the resulting connection - /// state on `BrowserCore`. Returns `true` when at least one - /// update was applied so callers can `cx.notify()` accordingly. - pub(super) fn drain_sync_updates(&mut self) -> bool { - let mut latest_connection: Option = None; - let mut auth_changed = false; - let mut trigger_initial_sync = false; - let mut trigger_merged_upload = None; - while let Ok(update) = self.sync_inbox_rx.try_recv() { - match update { - SyncStateUpdate::SignedOut => { - latest_connection = Some(ely_domain::SyncConnectionState::SignedOut); - } - SyncStateUpdate::AwaitingDeviceApproval => { - latest_connection = - Some(ely_domain::SyncConnectionState::AwaitingDeviceApproval); - } - SyncStateUpdate::RemoteSnapshot { bytes, logical_clock } => { - if let ShellState::Ready(core) = &mut self.state { - match core.apply_sync_snapshot_bytes(&bytes) { - Ok(summary) => { - tracing::info!( - target: "ely::sync", - imported = summary.imported(), - updated = summary.updated(), - skipped = summary.skipped(), - "remote snapshot applied", - ); - trigger_merged_upload = Some(logical_clock); - } - Err(error) => { - latest_connection = - Some(ely_domain::SyncConnectionState::SyncError { - message: error.to_string(), - }); - } - } - } - } - SyncStateUpdate::SyncReady { last_synced_at_secs } => { - latest_connection = - Some(ely_domain::SyncConnectionState::SyncReady { last_synced_at_secs }); - } - SyncStateUpdate::SyncError { message } => { - latest_connection = - Some(ely_domain::SyncConnectionState::SyncError { message }); - } - SyncStateUpdate::AuthOtpSent { email } => { - self.auth_flow_phase = auth::AuthFlowPhase::AwaitingOtp { email }; - auth_changed = true; - } - SyncStateUpdate::AuthSucceeded { email } => { - self.auth_flow_phase = auth::AuthFlowPhase::Idle; - latest_connection = Some(ely_domain::SyncConnectionState::SignedIn); - trigger_initial_sync = true; - tracing::info!(target: "ely::sync", email = %email, "email OTP sign-in succeeded"); - auth_changed = true; - } - SyncStateUpdate::AuthError { email, message } => { - self.auth_flow_phase = auth::AuthFlowPhase::Error { email, message }; - auth_changed = true; - } - } - } - if let (Some(state), ShellState::Ready(core)) = (latest_connection, &mut self.state) { - core.set_sync_connection_state(state); - } - if trigger_initial_sync { - self.trigger_cloud_sync_upload(); - } - let merged_upload_requested = trigger_merged_upload.is_some(); - if let Some(logical_clock_floor) = trigger_merged_upload { - self.trigger_cloud_sync_upload_after_remote(logical_clock_floor); - } - auth_changed || trigger_initial_sync || merged_upload_requested - } - fn focus_command_mode(&mut self, window: &mut Window, cx: &mut Context) { if let ShellState::Ready(core) = &mut self.state { core.set_command_query(">"); @@ -340,6 +271,7 @@ impl ElyShell { && core.select_tab(tab_id).is_ok() { self.sync_address_input(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -349,6 +281,7 @@ impl ElyShell { && core.select_space(space_id).is_ok() { self.sync_address_input(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -363,6 +296,7 @@ impl ElyShell { && core.select_profile(profile_id).is_ok() { self.sync_address_input(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -372,6 +306,7 @@ impl ElyShell { && core.select_next_tab().is_ok() { self.sync_address_input(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -381,6 +316,7 @@ impl ElyShell { && core.select_previous_tab().is_ok() { self.sync_address_input(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -390,6 +326,7 @@ impl ElyShell { && core.close_active_tab().is_ok() { self.sync_address_input(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -399,6 +336,7 @@ impl ElyShell { && core.restore_last_archived_tab().is_ok() { self.sync_address_input(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -413,6 +351,7 @@ impl ElyShell { && core.restore_archived_tab(tab_id).is_ok() { self.sync_address_input(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -421,6 +360,7 @@ impl ElyShell { if let ShellState::Ready(core) = &mut self.state && core.toggle_active_tab_favorite().is_ok() { + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -429,6 +369,7 @@ impl ElyShell { if let ShellState::Ready(core) = &mut self.state && core.toggle_active_tab_pinned().is_ok() { + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -437,6 +378,7 @@ impl ElyShell { if let ShellState::Ready(core) = &mut self.state && core.zoom_active_tab_in().is_ok() { + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -445,6 +387,7 @@ impl ElyShell { if let ShellState::Ready(core) = &mut self.state && core.zoom_active_tab_out().is_ok() { + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -453,6 +396,7 @@ impl ElyShell { if let ShellState::Ready(core) = &mut self.state && core.reset_active_tab_zoom().is_ok() { + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -467,7 +411,7 @@ fn start_external_web_surface_timer(cx: &mut Context) { }; Timer::after(delay).await; let result = shell.update(cx, |shell, cx| { - if shell.tick_external_web_surfaces() { + if shell.tick_external_web_surfaces(cx) { cx.notify(); } }); diff --git a/crates/ely_app/src/shell/navigation.rs b/crates/ely_app/src/shell/navigation.rs index e2f3a97..e27bb4a 100644 --- a/crates/ely_app/src/shell/navigation.rs +++ b/crates/ely_app/src/shell/navigation.rs @@ -18,6 +18,7 @@ impl ElyShell { { self.sync_address_input(window, cx); self.focus_address_bar(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -74,6 +75,7 @@ impl ElyShell { core.open_tab(url); } self.sync_address_input(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -83,6 +85,7 @@ impl ElyShell { && core.navigate_active_tab_back().is_ok_and(|changed| changed) { self.sync_address_input(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -96,6 +99,7 @@ impl ElyShell { && core.navigate_active_tab_forward().is_ok_and(|changed| changed) { self.sync_address_input(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -108,6 +112,7 @@ impl ElyShell { core.open_tab(url); self.sync_address_input(window, cx); self.focus_address_bar(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } diff --git a/crates/ely_app/src/shell/settings_actions.rs b/crates/ely_app/src/shell/settings_actions.rs index 5c9bbd4..2eef35a 100644 --- a/crates/ely_app/src/shell/settings_actions.rs +++ b/crates/ely_app/src/shell/settings_actions.rs @@ -193,6 +193,7 @@ impl ElyShell { if let ShellState::Ready(core) = &mut self.state && core.set_profile_sync_policy(profile_id, sync_policy).is_ok() { + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -200,6 +201,7 @@ impl ElyShell { pub(super) fn reset_profile_sync_settings(&mut self, cx: &mut Context) { if let ShellState::Ready(core) = &mut self.state { core.reset_profile_sync_settings(); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -231,6 +233,7 @@ impl ElyShell { ) { if let ShellState::Ready(core) = &mut self.state { core.set_sync_object_policy(kind, policy); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -238,6 +241,7 @@ impl ElyShell { pub(super) fn reset_sync_settings(&mut self, cx: &mut Context) { if let ShellState::Ready(core) = &mut self.state { core.reset_sync_settings(); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -251,6 +255,13 @@ impl ElyShell { } fn trigger_cloud_sync_upload_with_clock_floor(&mut self, logical_clock_floor: Option) { + if self.sync_upload_in_flight { + self.sync_upload_scheduled = false; + self.queue_cloud_sync_upload(logical_clock_floor); + return; + } + self.sync_upload_scheduled = false; + let ShellState::Ready(core) = &self.state else { return; }; @@ -278,19 +289,19 @@ impl ElyShell { let tx = self.sync_inbox_tx.clone(); let thread_name = if logical_clock_floor.is_some() { "ely-sync-merge-upload" } else { "ely-sync-upload" }; - std::thread::Builder::new() - .name(thread_name.to_string()) - .spawn(move || { + self.sync_upload_in_flight = true; + if let Err(error) = + std::thread::Builder::new().name(thread_name.to_string()).spawn(move || { run_sync_upload(profile_dir, device_name, bytes, logical_clock_floor, tx) }) - .map(|_| ()) - .unwrap_or_else(|error| { - tracing::warn!( - target: "ely::sync", - error = %error, - "failed to spawn ely-sync-upload thread", - ); - }); + { + self.sync_upload_in_flight = false; + tracing::warn!( + target: "ely::sync", + error = %error, + "failed to spawn ely-sync-upload thread", + ); + } } pub(super) fn set_update_policy( diff --git a/crates/ely_app/src/shell/sync_state.rs b/crates/ely_app/src/shell/sync_state.rs index c6291b1..243f320 100644 --- a/crates/ely_app/src/shell/sync_state.rs +++ b/crates/ely_app/src/shell/sync_state.rs @@ -1,6 +1,11 @@ -use std::path::Path; +use std::{path::Path, time::Duration}; -use super::{ElyShell, ShellState}; +use ely_domain::SyncConnectionState; +use gpui::{Context, Timer}; + +use super::{ElyShell, ShellState, auth}; + +const CLOUD_SYNC_UPLOAD_DEBOUNCE: Duration = Duration::from_millis(750); /// Messages the off-thread sync workers push back to the shell so /// `SyncConnectionState` on `BrowserCore` and the in-flight auth @@ -34,6 +39,76 @@ pub(crate) const fn sync_platform_label() -> &'static str { } impl ElyShell { + pub(crate) fn schedule_cloud_sync_upload(&mut self, cx: &mut Context) { + if !self.can_schedule_cloud_sync_upload() { + return; + } + if self.sync_upload_in_flight { + self.queue_cloud_sync_upload(None); + return; + } + if self.sync_upload_scheduled { + return; + } + + self.sync_upload_scheduled = true; + cx.spawn(async move |shell, cx| { + Timer::after(CLOUD_SYNC_UPLOAD_DEBOUNCE).await; + let _ = shell.update(cx, |shell, _| { + if shell.sync_upload_scheduled { + shell.sync_upload_scheduled = false; + shell.trigger_cloud_sync_upload(); + } + }); + }) + .detach(); + } + + pub(super) fn queue_cloud_sync_upload(&mut self, logical_clock_floor: Option) { + self.sync_upload_pending = true; + if let Some(floor) = logical_clock_floor { + self.sync_upload_pending_logical_clock_floor = Some( + self.sync_upload_pending_logical_clock_floor + .map_or(floor, |current| current.max(floor)), + ); + } + } + + fn trigger_pending_cloud_sync_upload(&mut self) -> bool { + if !self.sync_upload_pending { + return false; + } + + self.sync_upload_pending = false; + let logical_clock_floor = self.sync_upload_pending_logical_clock_floor.take(); + match logical_clock_floor { + Some(floor) => self.trigger_cloud_sync_upload_after_remote(floor), + None => self.trigger_cloud_sync_upload(), + } + true + } + + fn clear_pending_cloud_sync_upload(&mut self) { + self.sync_upload_pending = false; + self.sync_upload_pending_logical_clock_floor = None; + } + + fn can_schedule_cloud_sync_upload(&self) -> bool { + let ShellState::Ready(core) = &self.state else { + return false; + }; + let Some(snapshot) = core.snapshot().ok() else { + return false; + }; + matches!( + snapshot.sync_status.connection(), + SyncConnectionState::SignedIn + | SyncConnectionState::AwaitingDeviceApproval + | SyncConnectionState::SyncReady { .. } + | SyncConnectionState::SyncError { .. } + ) + } + /// Inspect the on-disk bearer token and seed `SyncConnectionState` /// so the Sync settings page reads the startup state on first render. pub(super) fn probe_initial_sync_state(&mut self) -> bool { @@ -62,6 +137,97 @@ impl ElyShell { core.set_sync_connection_state(state); bearer_present } + + /// Drain any sync upload outcomes the off-thread worker pushed + /// since the previous tick and stamp the resulting connection + /// state on `BrowserCore`. Returns `true` when at least one + /// update was applied so callers can `cx.notify()` accordingly. + pub(super) fn drain_sync_updates(&mut self) -> bool { + let mut latest_connection: Option = None; + let mut auth_changed = false; + let mut trigger_initial_sync = false; + let mut trigger_merged_upload = None; + let mut upload_finished = false; + while let Ok(update) = self.sync_inbox_rx.try_recv() { + match update { + SyncStateUpdate::SignedOut => { + latest_connection = Some(SyncConnectionState::SignedOut); + upload_finished = true; + } + SyncStateUpdate::AwaitingDeviceApproval => { + latest_connection = Some(SyncConnectionState::AwaitingDeviceApproval); + upload_finished = true; + } + SyncStateUpdate::RemoteSnapshot { bytes, logical_clock } => { + upload_finished = true; + if let ShellState::Ready(core) = &mut self.state { + match core.apply_sync_snapshot_bytes(&bytes) { + Ok(summary) => { + tracing::info!( + target: "ely::sync", + imported = summary.imported(), + updated = summary.updated(), + skipped = summary.skipped(), + "remote snapshot applied", + ); + trigger_merged_upload = Some(logical_clock); + } + Err(error) => { + latest_connection = Some(SyncConnectionState::SyncError { + message: error.to_string(), + }); + } + } + } + } + SyncStateUpdate::SyncReady { last_synced_at_secs } => { + latest_connection = + Some(SyncConnectionState::SyncReady { last_synced_at_secs }); + upload_finished = true; + } + SyncStateUpdate::SyncError { message } => { + latest_connection = Some(SyncConnectionState::SyncError { message }); + upload_finished = true; + } + SyncStateUpdate::AuthOtpSent { email } => { + self.auth_flow_phase = auth::AuthFlowPhase::AwaitingOtp { email }; + auth_changed = true; + } + SyncStateUpdate::AuthSucceeded { email } => { + self.auth_flow_phase = auth::AuthFlowPhase::Idle; + latest_connection = Some(SyncConnectionState::SignedIn); + trigger_initial_sync = true; + tracing::info!(target: "ely::sync", email = %email, "email OTP sign-in succeeded"); + auth_changed = true; + } + SyncStateUpdate::AuthError { email, message } => { + self.auth_flow_phase = auth::AuthFlowPhase::Error { email, message }; + auth_changed = true; + } + } + } + + let connection_changed = latest_connection.is_some(); + if let (Some(state), ShellState::Ready(core)) = (latest_connection, &mut self.state) { + core.set_sync_connection_state(state); + } + if upload_finished { + self.sync_upload_in_flight = false; + } + if trigger_initial_sync { + self.trigger_cloud_sync_upload(); + } + + let merged_upload_requested = trigger_merged_upload.is_some(); + if let Some(logical_clock_floor) = trigger_merged_upload { + self.clear_pending_cloud_sync_upload(); + self.trigger_cloud_sync_upload_after_remote(logical_clock_floor); + } else if upload_finished { + self.trigger_pending_cloud_sync_upload(); + } + + auth_changed || trigger_initial_sync || merged_upload_requested || connection_changed + } } fn bearer_token_file_present(path: &Path) -> bool { diff --git a/crates/ely_app/src/shell/tab_lifecycle.rs b/crates/ely_app/src/shell/tab_lifecycle.rs index 1235434..465f1a8 100644 --- a/crates/ely_app/src/shell/tab_lifecycle.rs +++ b/crates/ely_app/src/shell/tab_lifecycle.rs @@ -15,6 +15,7 @@ impl ElyShell { { self.sync_address_input(window, cx); self.focus_address_bar(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -30,6 +31,7 @@ impl ElyShell { { self.sync_address_input(window, cx); self.focus_address_bar(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } @@ -50,6 +52,7 @@ impl ElyShell { && core.close_tab(tab_id).is_ok() { self.sync_address_input(window, cx); + self.schedule_cloud_sync_upload(cx); cx.notify(); } } diff --git a/crates/ely_app/src/shell/web_surface_controller.rs b/crates/ely_app/src/shell/web_surface_controller.rs index 2058427..db46935 100644 --- a/crates/ely_app/src/shell/web_surface_controller.rs +++ b/crates/ely_app/src/shell/web_surface_controller.rs @@ -45,7 +45,7 @@ impl ElyShell { } } - pub(super) fn tick_external_web_surfaces(&mut self) -> bool { + pub(super) fn tick_external_web_surfaces(&mut self, cx: &mut Context) -> bool { let (visible_tab_ids, open_tab_ids, visible_tabs) = match &self.state { super::ShellState::Ready(core) => { let visible_tabs = core.visible_content_tabs().unwrap_or_else(|_| Vec::new()); @@ -65,6 +65,9 @@ impl ElyShell { metadata_changed |= self.apply_web_surface_page_metadata(metadata); } let sync_changed = self.drain_sync_updates(); + if url_changed || metadata_changed { + self.schedule_cloud_sync_upload(cx); + } result.changed || url_changed || metadata_changed || sync_changed } @@ -79,7 +82,7 @@ impl ElyShell { } fn flush_external_web_surface_tick(&mut self, cx: &mut Context) { - if self.tick_external_web_surfaces() { + if self.tick_external_web_surfaces(cx) { cx.notify(); } }