From bb38b1a28f941721343de29ccbc8aeab6506f897 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pascal=20Andr=C3=A9?= Date: Sat, 5 Sep 2026 15:07:18 +0200 Subject: [PATCH 1/2] fix(tauri): coalesce window state flushes Replace the per-move and per-resize thread spawn with one lazily started, bounded debounce worker. Window bounds are still captured immediately, while disk persistence is coalesced after 250 ms so native event handling cannot accumulate hundreds of flush threads behind the client-state write lock. Keep at most one trailing request while a write is active and drain pending work before releasing cross-host ownership. Pass AppHandle values through the queue instead of retaining one while idle so the worker does not extend the application lifetime. Add regression coverage for event bursts, requests arriving during an active flush, and prompt draining during shutdown. Fixes #676. --- .../tauri-app/src-tauri/src/client_state.rs | 15 ++- .../src-tauri/src/client_state/window.rs | 18 +--- .../src/client_state/window_flush.rs | 96 +++++++++++++++++++ .../src/client_state/window_flush_tests.rs | 92 ++++++++++++++++++ 4 files changed, 201 insertions(+), 20 deletions(-) create mode 100644 packages/tauri-app/src-tauri/src/client_state/window_flush.rs create mode 100644 packages/tauri-app/src-tauri/src/client_state/window_flush_tests.rs diff --git a/packages/tauri-app/src-tauri/src/client_state.rs b/packages/tauri-app/src-tauri/src/client_state.rs index 3b5bfcf8a..3e4d3df4e 100644 --- a/packages/tauri-app/src-tauri/src/client_state.rs +++ b/packages/tauri-app/src-tauri/src/client_state.rs @@ -6,6 +6,9 @@ mod navigation; mod partitions; mod process; mod window; +mod window_flush; +#[cfg(test)] +mod window_flush_tests; #[doc(hidden)] pub use commands::{ @@ -35,7 +38,6 @@ use std::collections::{HashMap, HashSet}; use std::fs; use std::io::Write; use std::path::{Path, PathBuf}; -use std::sync::atomic::AtomicU64; use std::sync::Mutex; use std::time::{Duration, Instant}; use tauri::{AppHandle, Emitter, Manager}; @@ -72,7 +74,7 @@ pub struct ClientState { state: Mutex, zoom_levels: Mutex>, write_lock: Mutex<()>, - save_generation: AtomicU64, + window_flush: window_flush::WindowFlushScheduler, renderer_access: access::RendererAccess, ephemeral_windows: Mutex>, renderer_flush: RendererFlush, @@ -205,7 +207,7 @@ impl ClientState { state: Mutex::new(state), zoom_levels: Mutex::new(zoom_levels), write_lock: Mutex::new(()), - save_generation: AtomicU64::new(0), + window_flush: window_flush::WindowFlushScheduler::default(), renderer_access: access::RendererAccess::default(), ephemeral_windows: Mutex::new(HashSet::new()), renderer_flush: RendererFlush::default(), @@ -704,6 +706,12 @@ impl ClientState { Ok(()) } + fn schedule_window_flush(&self, app: &AppHandle) { + if let Err(error) = self.window_flush.schedule(app) { + eprintln!("[client-state] failed to schedule window-state flush: {error}"); + } + } + fn normal_writes_suppressed(&self, window_id: &str) -> Result { let state = self.state.lock().map_err(|err| err.to_string())?; Ok(state.unsupported_future_envelope || !state.record(window_id)?.writes_enabled) @@ -822,6 +830,7 @@ impl ClientState { } fn release_locks(&self) { + self.window_flush.stop(); // Lock order fences takeover until root publication and partition GC leave write_lock. let _write = self .write_lock diff --git a/packages/tauri-app/src-tauri/src/client_state/window.rs b/packages/tauri-app/src-tauri/src/client_state/window.rs index 4d6bc662c..abd983755 100644 --- a/packages/tauri-app/src-tauri/src/client_state/window.rs +++ b/packages/tauri-app/src-tauri/src/client_state/window.rs @@ -1,16 +1,12 @@ use super::ClientState; use serde::{Deserialize, Serialize}; use serde_json::Value; -use std::sync::atomic::Ordering; -use std::time::Duration; use tauri::{AppHandle, Manager, PhysicalPosition, PhysicalSize, WindowEvent}; const MIN_WINDOW_WIDTH: i32 = 800; const MIN_WINDOW_HEIGHT: i32 = 600; const MIN_ZOOM_LEVEL: f64 = 0.25; pub(super) const MAX_ZOOM_LEVEL: f64 = 5.0; -const SAVE_DEBOUNCE: Duration = Duration::from_millis(250); - pub const DEFAULT_ZOOM_LEVEL: f64 = 1.0; #[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] @@ -254,19 +250,7 @@ fn schedule_flush(app: &AppHandle) { let Some(client_state) = app.try_state::() else { return; }; - let generation = client_state.save_generation.fetch_add(1, Ordering::SeqCst) + 1; - let app = app.clone(); - std::thread::spawn(move || { - std::thread::sleep(SAVE_DEBOUNCE); - let Some(client_state) = app.try_state::() else { - return; - }; - if client_state.save_generation.load(Ordering::SeqCst) == generation { - if let Err(err) = client_state.flush() { - eprintln!("[client-state] failed to save window state: {err}"); - } - } - }); + client_state.schedule_window_flush(app); } #[cfg(windows)] diff --git a/packages/tauri-app/src-tauri/src/client_state/window_flush.rs b/packages/tauri-app/src-tauri/src/client_state/window_flush.rs new file mode 100644 index 000000000..dd3733497 --- /dev/null +++ b/packages/tauri-app/src-tauri/src/client_state/window_flush.rs @@ -0,0 +1,96 @@ +use super::ClientState; +use std::sync::{ + mpsc::{sync_channel, Receiver, RecvTimeoutError, SyncSender, TrySendError}, + Mutex, +}; +use std::thread::{self, JoinHandle}; +use std::time::Duration; +use tauri::{AppHandle, Manager}; + +const SAVE_DEBOUNCE: Duration = Duration::from_millis(250); + +#[derive(Default)] +struct SchedulerState { + sender: Option>, + worker: Option>, + stopped: bool, +} + +#[derive(Default)] +pub(super) struct WindowFlushScheduler { + state: Mutex, +} + +impl WindowFlushScheduler { + pub(super) fn schedule(&self, app: &AppHandle) -> Result<(), String> { + let mut state = self.state.lock().map_err(|error| error.to_string())?; + if state.stopped { + return Ok(()); + } + if state.sender.is_none() { + // Keep at most one wakeup queued while the single worker is busy. + let (sender, receiver): (SyncSender, Receiver) = sync_channel(1); + let worker = thread::Builder::new() + .name("client-state-window-flush".to_string()) + .spawn(move || { + run_debounced(receiver, SAVE_DEBOUNCE, |worker_app| { + let Some(client_state) = worker_app.try_state::() else { + return; + }; + if let Err(error) = client_state.flush() { + eprintln!("[client-state] failed to save window state: {error}"); + } + }); + }) + .map_err(|error| format!("failed to start window-state flush worker: {error}"))?; + state.sender = Some(sender); + state.worker = Some(worker); + } + + match state + .sender + .as_ref() + .expect("initialized sender") + .try_send(app.clone()) + { + Ok(()) | Err(TrySendError::Full(_)) => Ok(()), + Err(TrySendError::Disconnected(_)) => { + Err("window-state flush worker disconnected".to_string()) + } + } + } + + pub(super) fn stop(&self) { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + state.stopped = true; + // Disconnect wakes the worker and makes it drain a pending flush before ownership release. + drop(state.sender.take()); + if let Some(worker) = state.worker.take() { + if worker.join().is_err() { + eprintln!("[client-state] window-state flush worker panicked"); + } + } + } +} + +pub(super) fn run_debounced( + receiver: Receiver, + debounce: Duration, + mut flush: impl FnMut(T), +) { + while let Ok(mut request) = receiver.recv() { + loop { + match receiver.recv_timeout(debounce) { + Ok(next) => request = next, + Err(RecvTimeoutError::Timeout) => { + flush(request); + break; + } + Err(RecvTimeoutError::Disconnected) => { + flush(request); + return; + } + } + } + } +} diff --git a/packages/tauri-app/src-tauri/src/client_state/window_flush_tests.rs b/packages/tauri-app/src-tauri/src/client_state/window_flush_tests.rs new file mode 100644 index 000000000..680b344b1 --- /dev/null +++ b/packages/tauri-app/src-tauri/src/client_state/window_flush_tests.rs @@ -0,0 +1,92 @@ +use super::window_flush::run_debounced; +use std::sync::mpsc::{self, sync_channel, RecvTimeoutError}; +use std::thread; +use std::time::Duration; + +#[test] +fn burst_requests_coalesce_to_one_flush() { + let (sender, receiver) = sync_channel(1); + let (flushed_sender, flushed_receiver) = mpsc::channel(); + let worker = thread::spawn(move || { + run_debounced(receiver, Duration::from_millis(30), |()| { + flushed_sender.send(()).unwrap(); + }); + }); + + sender.try_send(()).unwrap(); + for _ in 0..256 { + let _ = sender.try_send(()); + } + flushed_receiver + .recv_timeout(Duration::from_secs(1)) + .unwrap(); + assert_eq!( + flushed_receiver.recv_timeout(Duration::from_millis(80)), + Err(RecvTimeoutError::Timeout) + ); + + drop(sender); + worker.join().unwrap(); +} + +#[test] +fn request_during_flush_produces_one_trailing_flush() { + let (sender, receiver) = sync_channel(1); + let (entered_sender, entered_receiver) = mpsc::channel(); + let (release_sender, release_receiver) = mpsc::channel(); + let worker = thread::spawn(move || { + let mut calls = 0; + run_debounced(receiver, Duration::from_millis(20), |()| { + calls += 1; + entered_sender.send(calls).unwrap(); + if calls == 1 { + release_receiver + .recv_timeout(Duration::from_secs(1)) + .unwrap(); + } + }); + }); + + sender.try_send(()).unwrap(); + assert_eq!( + entered_receiver + .recv_timeout(Duration::from_secs(1)) + .unwrap(), + 1 + ); + for _ in 0..256 { + let _ = sender.try_send(()); + } + release_sender.send(()).unwrap(); + assert_eq!( + entered_receiver + .recv_timeout(Duration::from_secs(1)) + .unwrap(), + 2 + ); + assert_eq!( + entered_receiver.recv_timeout(Duration::from_millis(60)), + Err(RecvTimeoutError::Timeout) + ); + + drop(sender); + worker.join().unwrap(); +} + +#[test] +fn disconnect_drains_a_pending_request_without_waiting_for_debounce() { + let (sender, receiver) = sync_channel(1); + let (flushed_sender, flushed_receiver) = mpsc::channel(); + let worker = thread::spawn(move || { + run_debounced(receiver, Duration::from_secs(10), |()| { + flushed_sender.send(()).unwrap(); + }); + }); + + sender.try_send(()).unwrap(); + drop(sender); + flushed_receiver + .recv_timeout(Duration::from_secs(1)) + .unwrap(); + worker.join().unwrap(); +} From 727197d55a00ce335ecb3a059eafe593d744a46f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pascal=20Andr=C3=A9?= Date: Sun, 6 Sep 2026 15:08:47 +0200 Subject: [PATCH 2/2] fix(tauri): decouple native window captures from disk persistence Complete #676's worker-only fix by reading native geometry before any client-state mutex and queueing only the latest capture per known window. Move/resize/DPI and zoom handlers no longer wait on atomic publication, and restore seeds clamped normal bounds before maximization. Ownership validation also leaves the in-memory mutex free during disk reads. Merge captures under the serialized writer before rollback snapshots. Preserve admission during tentative clear, disable and removal so failed publication cannot lose concurrent events; discard those speculative events after a successful destructive commit. Close admission before draining, retry previously merged but unpublished captures, and retain cross-host fencing until final persistence completes. Keep a single lazy worker and capacity-one queue, but consume callback ownership while idle and release the scheduler mutex before joining. Concurrent stop callers wait for the same drain while late UI submissions return promptly. Add sixteen Rust regressions beyond the original three scheduler tests, an Electron parity test, and an independent macOS ARM64 CI test gate. Windows validation passes all 158 Rust tests at default parallelism, all 190 Electron tests, UI/Electron typechecks and a release build. The same source-extraction harness that reproduces the old disk/UI lock failures now passes all five assertions over ten repetitions. Linux CI still depends on the separate #671 registry fix; an interactive macOS smoke test remains necessary. --- .github/workflows/pr-build.yml | 34 ++ .../electron/main/window-state.test.ts | 47 ++ .../tauri-app/src-tauri/src/client_state.rs | 68 ++- .../src-tauri/src/client_state/window.rs | 87 ++- .../src/client_state/window_flush.rs | 61 +- .../src/client_state/window_flush_tests.rs | 105 +++- .../src/client_state/window_updates.rs | 204 +++++++ .../src/client_state/window_updates_tests.rs | 545 ++++++++++++++++++ 8 files changed, 1059 insertions(+), 92 deletions(-) create mode 100644 packages/tauri-app/src-tauri/src/client_state/window_updates.rs create mode 100644 packages/tauri-app/src-tauri/src/client_state/window_updates_tests.rs diff --git a/.github/workflows/pr-build.yml b/.github/workflows/pr-build.yml index 76fa00d00..279576862 100644 --- a/.github/workflows/pr-build.yml +++ b/.github/workflows/pr-build.yml @@ -50,6 +50,7 @@ jobs: - authorize - tests - tests-tauri-windows + - tests-tauri-macos if: ${{ needs.authorize.outputs.allowed == 'true' && !github.event.pull_request.draft }} uses: ./.github/workflows/build-and-upload.yml with: @@ -217,3 +218,36 @@ jobs: - name: Test Tauri crate on Windows working-directory: packages/tauri-app/src-tauri run: cargo test --locked -- --test-threads=1 + + # Exercise the window persistence regressions on the architecture reported in + # #676, independently of the Linux server gate. Packaging alone cannot test them. + tests-tauri-macos: + needs: authorize + if: ${{ needs.authorize.outputs.allowed == 'true' && !github.event.pull_request.draft }} + runs-on: macos-26 + steps: + - name: Checkout + uses: actions/checkout@v4 + with: + ref: ${{ github.event.pull_request.head.sha }} + + - name: Setup Node + uses: actions/setup-node@v4 + with: + node-version: 22 + cache: npm + + - name: Setup Rust + uses: dtolnay/rust-toolchain@stable + + - name: Install dependencies + run: npm ci + + - name: Prepare Tauri test resources + run: >- + npm run dev:prep --workspace @codenomad/tauri-app && + node -e "require('fs').mkdirSync('packages/tauri-app/src-tauri/resources/server',{recursive:true})" + + - name: Test Tauri crate on macOS ARM64 + working-directory: packages/tauri-app/src-tauri + run: cargo test --locked -- --test-threads=4 diff --git a/packages/electron-app/electron/main/window-state.test.ts b/packages/electron-app/electron/main/window-state.test.ts index 94c6a761a..cf2d90e78 100644 --- a/packages/electron-app/electron/main/window-state.test.ts +++ b/packages/electron-app/electron/main/window-state.test.ts @@ -1,4 +1,5 @@ import assert from "node:assert/strict" +import { EventEmitter } from "node:events" import test from "node:test" import { clampWindowBounds, installWindowZoomInput, normalizeNativeWindowState, normalizeZoomFactor, restoreWindowState, WindowStateTracker } from "./window-state" import type { BrowserWindow } from "electron" @@ -6,6 +7,52 @@ import type { ClientStateManager } from "./client-state" const primaryDisplay = { x: 0, y: 0, width: 1920, height: 1080 } +test("move/resize bursts debounce and final flush preserves pre-maximize bounds", async (t) => { + t.mock.timers.enable({ apis: ["setTimeout"] }) + const events = new EventEmitter() + let x = 0 + let maximized = false + const window = Object.assign(events, { + isDestroyed: () => false, + getPosition: () => [x, 20], + getContentSize: () => [1200, 800], + isMaximized: () => maximized, + isFullScreen: () => false, + webContents: Object.assign(new EventEmitter(), { + isDestroyed: () => false, + getZoomFactor: () => 1.25, + }), + }) as unknown as BrowserWindow + const saved: unknown[] = [] + const manager = { + activeWindowId: "window-a", + saveWindowState: async (state: unknown, id: string) => { saved.push({ state, id }); return true }, + flush: async () => undefined, + } as unknown as ClientStateManager + const tracker = new WindowStateTracker(window, manager) + for (x = 1; x <= 1000; x++) { + events.emit("move") + events.emit("resize") + } + assert.equal(saved.length, 0) + maximized = true + events.emit("maximize") + t.mock.timers.tick(250) + assert.deepEqual(saved, [{ id: "window-a", state: { + bounds: { x: 1000, y: 20, width: 1200, height: 800 }, maximized: true, fullscreen: false, zoomFactor: 1.25, + } }]) + maximized = false + events.emit("move") + await tracker.flush() + assert.equal(saved.length, 2) + t.mock.timers.tick(1000) + assert.equal(saved.length, 2, "explicit flush cancels the delayed write") + events.emit("resize") + events.emit("closed") + t.mock.timers.tick(1000) + assert.equal(saved.length, 2, "closed windows leave no pending timer") +}) + test("normalizes persisted window state", () => { assert.equal(normalizeNativeWindowState({ bounds: { x: 0, y: 0, width: Number.NaN, height: 900 }, maximized: false, fullscreen: false, zoomFactor: 1 }), undefined) assert.deepEqual(clampWindowBounds({ x: 4000, y: 2000, width: 1400, height: 900 }, [primaryDisplay]), { x: 520, y: 180, width: 1400, height: 900 }) diff --git a/packages/tauri-app/src-tauri/src/client_state.rs b/packages/tauri-app/src-tauri/src/client_state.rs index 3e4d3df4e..4640421e0 100644 --- a/packages/tauri-app/src-tauri/src/client_state.rs +++ b/packages/tauri-app/src-tauri/src/client_state.rs @@ -7,8 +7,7 @@ mod partitions; mod process; mod window; mod window_flush; -#[cfg(test)] -mod window_flush_tests; +mod window_updates; #[doc(hidden)] pub use commands::{ @@ -75,6 +74,7 @@ pub struct ClientState { zoom_levels: Mutex>, write_lock: Mutex<()>, window_flush: window_flush::WindowFlushScheduler, + pending_windows: Mutex, renderer_access: access::RendererAccess, ephemeral_windows: Mutex>, renderer_flush: RendererFlush, @@ -208,6 +208,7 @@ impl ClientState { zoom_levels: Mutex::new(zoom_levels), write_lock: Mutex::new(()), window_flush: window_flush::WindowFlushScheduler::default(), + pending_windows: Mutex::new(window_updates::WindowCaptures::default()), renderer_access: access::RendererAccess::default(), ephemeral_windows: Mutex::new(HashSet::new()), renderer_flush: RendererFlush::default(), @@ -404,6 +405,9 @@ impl ClientState { } fn load_window(&self, window_id: &str) -> Result { + // Ownership validation reads the election files. Do not make native + // capture wait for that I/O through the shared in-memory state mutex. + let is_primary = self.is_primary(); let state = self.state.lock().map_err(|err| err.to_string())?; let record = match state.record(window_id) { Ok(record) => record, @@ -423,7 +427,6 @@ impl ClientState { } Err(error) => return Err(error), }; - let is_primary = self.is_primary(); Ok(ClientStateLoadResult { is_primary, restore_enabled: if is_primary || !self.process.is_registered() { @@ -701,13 +704,25 @@ impl ClientState { .map_err(|err| err.to_string())? .unsupported_future_envelope; if self.is_primary() && !unsupported { + { + let mut state = self.state.lock().map_err(|err| err.to_string())?; + self.apply_window_captures(&mut state)?; + } self.write_current_state()?; } Ok(()) } fn schedule_window_flush(&self, app: &AppHandle) { - if let Err(error) = self.window_flush.schedule(app) { + let app = app.clone(); + if let Err(error) = self.window_flush.schedule(move || { + // Only persist captured data; never dispatch native getters from this worker. + if let Some(state) = app.try_state::() { + if let Err(error) = state.flush() { + eprintln!("[client-state] failed to save window state: {error}"); + } + } + }) { eprintln!("[client-state] failed to schedule window-state flush: {error}"); } } @@ -736,29 +751,35 @@ impl ClientState { (self.write_state)(&self.state_path, &bytes, &|| { self.is_primary() && replacement_valid() })?; + // New events may already be queued, but only this serialized writer can + // have merged captures into the state we just published. + self.window_captures_published(); Ok(()) } fn mutate_and_write( &self, - _window_id: &str, + window_id: &str, mutate: impl FnOnce(&mut PersistedClientState) -> Result<(), String>, replacement_valid: &dyn Fn() -> bool, ) -> Result { let previous_state = { let mut state = self.state.lock().map_err(|err| err.to_string())?; + // Preserve captures in rollback; arrivals during I/O stay in the mailbox. + self.apply_window_captures(&mut state)?; let previous = state.clone(); mutate(&mut state)?; + self.preserve_window_capture_policy(&previous, window_id); previous }; - match self.write_current_state_guarded(replacement_valid) { - Ok(()) => Ok(true), - Err(err) => { - *self.state.lock().map_err(|lock_err| lock_err.to_string())? = previous_state; - Err(err) - } + let result = self.write_current_state_guarded(replacement_valid); + let mut state = self.state.lock().map_err(|error| error.to_string())?; + if result.is_err() { + *state = previous_state; } + self.finish_window_capture_policy(&state, window_id); + result.map(|()| true) } pub(crate) fn add_window(&self, window_id: String) -> Result { @@ -807,35 +828,50 @@ impl ClientState { if !self.is_primary() { return Ok(false); } - let mut zoom_levels = self.zoom_levels.lock().map_err(|err| err.to_string())?; let previous = { let mut state = self.state.lock().map_err(|err| err.to_string())?; if state.unsupported_future_envelope { return Ok(false); } + self.apply_window_captures(&mut state)?; let previous = state.clone(); if !state.remove_window(window_id)? { return Ok(false); } + self.preserve_window_capture_policy(&previous, window_id); previous }; - if let Err(err) = self.write_current_state() { - *self.state.lock().map_err(|lock_err| lock_err.to_string())? = previous; - return Err(err); + let result = self.write_current_state(); + { + let mut state = self.state.lock().map_err(|error| error.to_string())?; + if result.is_err() { + *state = previous; + } + self.finish_window_capture_policy(&state, window_id); } + result?; self.renderer_access.remove(window_id); - zoom_levels.remove(window_id); + self.zoom_levels + .lock() + .map_err(|err| err.to_string())? + .remove(window_id); self.collect_partitions(&|| true); Ok(true) } fn release_locks(&self) { + self.stop_window_captures(); self.window_flush.stop(); // Lock order fences takeover until root publication and partition GC leave write_lock. let _write = self .write_lock .lock() .unwrap_or_else(|err| err.into_inner()); + // Also drain a capture admitted before stop whose wakeup wasn't sent yet, + // or retry geometry merged by a worker whose publication failed. + if let Err(error) = self.flush_pending_window_captures() { + eprintln!("[client-state] failed to drain final window state: {error}"); + } self.process.release_locks(); } diff --git a/packages/tauri-app/src-tauri/src/client_state/window.rs b/packages/tauri-app/src-tauri/src/client_state/window.rs index abd983755..5d851407a 100644 --- a/packages/tauri-app/src-tauri/src/client_state/window.rs +++ b/packages/tauri-app/src-tauri/src/client_state/window.rs @@ -1,3 +1,4 @@ +use super::window_updates::WindowGeometry; use super::ClientState; use serde::{Deserialize, Serialize}; use serde_json::Value; @@ -176,29 +177,25 @@ fn center_distance_squared(bounds: &WindowBounds, display: DisplayArea) -> i128 (bounds_x - display_x).pow(2) + (bounds_y - display_y).pow(2) } -fn capture_window_in_memory(app: &AppHandle, window_label: &str, window_id: &str, persisted: bool) { +fn capture_window_in_memory( + app: &AppHandle, + window_label: &str, + window_id: &str, + persisted: bool, +) -> bool { if !persisted { - return; + return false; } let Some(client_state) = app.try_state::() else { - return; - }; - let Ok(_write) = client_state.write_lock.lock() else { - return; + return false; }; - if !client_state.is_primary() { - return; - } - if client_state - .normal_writes_suppressed(window_id) - .unwrap_or(true) - { - return; - } let Some(window) = app.get_webview_window(window_label) else { - return; + return false; }; + client_state.capture_window_geometry(window_id, || read_window_geometry(&window)) +} +fn read_window_geometry(window: &tauri::WebviewWindow) -> WindowGeometry { let maximized = window.is_maximized().unwrap_or(false); let fullscreen = window.is_fullscreen().unwrap_or(false); let minimized = window.is_minimized().unwrap_or(false); @@ -222,27 +219,10 @@ fn capture_window_in_memory(app: &AppHandle, window_label: &str, window_id: &str } else { None }; - let zoom_factor = client_state - .zoom_levels - .lock() - .ok() - .and_then(|zoom| zoom.get(window_id).copied()) - .unwrap_or(DEFAULT_ZOOM_LEVEL); - let Ok(mut state) = client_state.state.lock() else { - return; - }; - let Ok(record) = state.record_mut(window_id) else { - return; - }; - let bounds = - current_bounds.or_else(|| record.window.as_ref().map(|window| window.bounds.clone())); - if let Some(bounds) = bounds { - record.window = Some(NativeWindowState { - bounds, - maximized, - fullscreen, - zoom_factor, - }); + WindowGeometry { + bounds: current_bounds, + maximized, + fullscreen, } } @@ -289,10 +269,7 @@ fn register_native_zoom_handler( zoom_levels.insert(window_id.clone(), normalized); drop(zoom_levels); - if client_state.is_primary() - && client_state.normal_writes_suppressed(&window_id).ok() == Some(false) - { - capture_window_in_memory(&callback_app, &window_label, &window_id, persisted); + if capture_window_in_memory(&callback_app, &window_label, &window_id, persisted) { schedule_flush(&callback_app); } Ok(()) @@ -375,11 +352,15 @@ pub fn setup_local_window( }; } } - if let Ok(mut state) = client_state.state.lock() { - if let Ok(record) = state.record_mut(window_id) { - record.window = Some(saved_window.clone()); - } - } + // Seed clamped normal bounds before maximizing, replacing any early + // native zoom callback's default geometry in the mailbox. + client_state.queue_window_capture( + window_id, + Some(saved_window.bounds.clone()), + saved_window.maximized, + saved_window.fullscreen, + saved_window.zoom_factor, + ); let _ = window.set_zoom(saved_window.zoom_factor); if saved_window.maximized { let _ = window.maximize(); @@ -392,7 +373,9 @@ pub fn setup_local_window( } } - capture_window_in_memory(app, window.label(), window_id, persisted); + if capture_window_in_memory(app, window.label(), window_id, persisted) { + schedule_flush(app); + } let app_handle = app.clone(); let window_label = window.label().to_string(); let window_id = window_id.to_string(); @@ -400,8 +383,9 @@ pub fn setup_local_window( WindowEvent::Resized(_) | WindowEvent::Moved(_) | WindowEvent::ScaleFactorChanged { .. } => { - capture_window_in_memory(&app_handle, &window_label, &window_id, persisted); - schedule_flush(&app_handle); + if capture_window_in_memory(&app_handle, &window_label, &window_id, persisted) { + schedule_flush(&app_handle); + } } _ => {} }); @@ -430,9 +414,8 @@ pub fn set_local_window_zoom(app: &AppHandle, window_label: &str, next_zoom: f64 .try_state::() .and_then(|windows| windows.record(window_label)) .is_some_and(|record| record.persisted); - capture_window_in_memory(app, window_label, &window_id, persisted); - if let Err(err) = client_state.flush() { - eprintln!("[client-state] failed to save zoom level: {err}"); + if capture_window_in_memory(app, window_label, &window_id, persisted) { + schedule_flush(app); } } diff --git a/packages/tauri-app/src-tauri/src/client_state/window_flush.rs b/packages/tauri-app/src-tauri/src/client_state/window_flush.rs index dd3733497..3f3ea83f6 100644 --- a/packages/tauri-app/src-tauri/src/client_state/window_flush.rs +++ b/packages/tauri-app/src-tauri/src/client_state/window_flush.rs @@ -1,47 +1,41 @@ -use super::ClientState; use std::sync::{ mpsc::{sync_channel, Receiver, RecvTimeoutError, SyncSender, TrySendError}, - Mutex, + Condvar, Mutex, }; use std::thread::{self, JoinHandle}; use std::time::Duration; -use tauri::{AppHandle, Manager}; const SAVE_DEBOUNCE: Duration = Duration::from_millis(250); +type Flush = Box; #[derive(Default)] struct SchedulerState { - sender: Option>, + sender: Option>, worker: Option>, + stopping: bool, stopped: bool, } #[derive(Default)] pub(super) struct WindowFlushScheduler { state: Mutex, + stopped: Condvar, } impl WindowFlushScheduler { - pub(super) fn schedule(&self, app: &AppHandle) -> Result<(), String> { + // Each callback persists the latest mailbox state, never reads native APIs. + // Consume it after flushing so no AppHandle is retained while the worker idles. + pub(super) fn schedule(&self, flush: impl FnOnce() + Send + 'static) -> Result<(), String> { let mut state = self.state.lock().map_err(|error| error.to_string())?; - if state.stopped { + if state.stopping || state.stopped { return Ok(()); } if state.sender.is_none() { // Keep at most one wakeup queued while the single worker is busy. - let (sender, receiver): (SyncSender, Receiver) = sync_channel(1); + let (sender, receiver) = sync_channel::(1); let worker = thread::Builder::new() .name("client-state-window-flush".to_string()) - .spawn(move || { - run_debounced(receiver, SAVE_DEBOUNCE, |worker_app| { - let Some(client_state) = worker_app.try_state::() else { - return; - }; - if let Err(error) = client_state.flush() { - eprintln!("[client-state] failed to save window state: {error}"); - } - }); - }) + .spawn(move || run_debounced(receiver, SAVE_DEBOUNCE, |flush| flush())) .map_err(|error| format!("failed to start window-state flush worker: {error}"))?; state.sender = Some(sender); state.worker = Some(worker); @@ -51,7 +45,7 @@ impl WindowFlushScheduler { .sender .as_ref() .expect("initialized sender") - .try_send(app.clone()) + .try_send(Box::new(flush)) { Ok(()) | Err(TrySendError::Full(_)) => Ok(()), Err(TrySendError::Disconnected(_)) => { @@ -61,18 +55,39 @@ impl WindowFlushScheduler { } pub(super) fn stop(&self) { - let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); - state.stopped = true; - // Disconnect wakes the worker and makes it drain a pending flush before ownership release. - drop(state.sender.take()); - if let Some(worker) = state.worker.take() { + let (sender, worker) = { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + while state.stopping { + state = self + .stopped + .wait(state) + .unwrap_or_else(|error| error.into_inner()); + } + if state.stopped { + return; + } + state.stopping = true; + (state.sender.take(), state.worker.take()) + }; + // Disconnect drains immediately. Never hold the scheduler mutex while + // joining: late UI event submissions must not wait for the worker's I/O. + drop(sender); + if let Some(worker) = worker { if worker.join().is_err() { eprintln!("[client-state] window-state flush worker panicked"); } } + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + state.stopping = false; + state.stopped = true; + self.stopped.notify_all(); } } +#[cfg(test)] +#[path = "window_flush_tests.rs"] +mod tests; + pub(super) fn run_debounced( receiver: Receiver, debounce: Duration, diff --git a/packages/tauri-app/src-tauri/src/client_state/window_flush_tests.rs b/packages/tauri-app/src-tauri/src/client_state/window_flush_tests.rs index 680b344b1..a567704d1 100644 --- a/packages/tauri-app/src-tauri/src/client_state/window_flush_tests.rs +++ b/packages/tauri-app/src-tauri/src/client_state/window_flush_tests.rs @@ -1,8 +1,111 @@ -use super::window_flush::run_debounced; +use super::*; use std::sync::mpsc::{self, sync_channel, RecvTimeoutError}; +use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, +}; use std::thread; use std::time::Duration; +#[test] +fn worker_and_queue_stay_bounded_and_idle_worker_releases_callback() { + let scheduler = WindowFlushScheduler::default(); + let (entered, entries) = mpsc::channel(); + let (release, releases) = mpsc::channel(); + let callback_owner = Arc::new(()); + let retained = Arc::downgrade(&callback_owner); + scheduler + .schedule(move || { + let _owner = callback_owner; + entered.send(thread::current().id()).unwrap(); + releases.recv_timeout(Duration::from_secs(5)).unwrap(); + }) + .unwrap(); + let worker_id = entries.recv_timeout(Duration::from_secs(5)).unwrap(); + let calls = Arc::new(AtomicUsize::new(0)); + let (trailing, trailing_calls) = mpsc::channel(); + for _ in 0..1_000 { + let calls = Arc::clone(&calls); + let trailing = trailing.clone(); + scheduler + .schedule(move || { + calls.fetch_add(1, Ordering::SeqCst); + trailing.send(thread::current().id()).unwrap(); + }) + .unwrap(); + } + assert_eq!(calls.load(Ordering::SeqCst), 0); + release.send(()).unwrap(); + assert_eq!( + trailing_calls.recv_timeout(Duration::from_secs(5)).unwrap(), + worker_id + ); + assert!( + retained.upgrade().is_none(), + "idle worker retained its AppHandle-like owner" + ); + scheduler.stop(); + assert_eq!(calls.load(Ordering::SeqCst), 1); + scheduler + .schedule(|| panic!("restarted after stop")) + .unwrap(); + scheduler.stop(); +} + +#[test] +fn concurrent_stop_waits_for_drain_but_late_ui_submission_does_not() { + let scheduler = Arc::new(WindowFlushScheduler::default()); + let (entered, entries) = mpsc::channel(); + let (release, releases) = mpsc::channel(); + scheduler + .schedule(move || { + entered.send(()).unwrap(); + releases.recv_timeout(Duration::from_secs(5)).unwrap(); + }) + .unwrap(); + entries.recv_timeout(Duration::from_secs(5)).unwrap(); + let stopping = Arc::clone(&scheduler); + let (done, completion) = mpsc::channel(); + let first = thread::spawn(move || { + stopping.stop(); + done.send(()).unwrap(); + }); + // Synchronize on the transition, not a sleep: stop has detached its worker. + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while !scheduler.state.lock().unwrap().stopping { + assert!(std::time::Instant::now() < deadline); + thread::yield_now(); + } + let second_scheduler = Arc::clone(&scheduler); + let (done, other_completion) = mpsc::channel(); + let second = thread::spawn(move || { + second_scheduler.stop(); + done.send(()).unwrap(); + }); + scheduler + .schedule(|| panic!("admitted after stop")) + .unwrap(); + assert!(completion.try_recv().is_err()); + assert!(other_completion.try_recv().is_err()); + release.send(()).unwrap(); + completion.recv_timeout(Duration::from_secs(5)).unwrap(); + other_completion + .recv_timeout(Duration::from_secs(5)) + .unwrap(); + first.join().unwrap(); + second.join().unwrap(); +} + +#[test] +fn stop_before_first_request_does_not_create_a_worker() { + let scheduler = WindowFlushScheduler::default(); + scheduler.stop(); + scheduler + .schedule(|| panic!("worker started after stop")) + .unwrap(); + assert!(scheduler.state.lock().unwrap().worker.is_none()); +} + #[test] fn burst_requests_coalesce_to_one_flush() { let (sender, receiver) = sync_channel(1); diff --git a/packages/tauri-app/src-tauri/src/client_state/window_updates.rs b/packages/tauri-app/src-tauri/src/client_state/window_updates.rs new file mode 100644 index 000000000..fe40ac5e5 --- /dev/null +++ b/packages/tauri-app/src-tauri/src/client_state/window_updates.rs @@ -0,0 +1,204 @@ +use super::{ + envelope::PersistedClientState, + window::{NativeWindowState, WindowBounds, DEFAULT_ZOOM_LEVEL}, + ClientState, +}; +use std::collections::HashMap; + +pub(super) struct WindowGeometry { + pub bounds: Option, + pub maximized: bool, + pub fullscreen: bool, +} + +#[derive(Default)] +pub(super) struct WindowCaptures { + latest: HashMap, + // A clear/disable/removal is tentative until publication. Keep admission + // based on its previous policy so a failed write cannot drop live events. + mutation: Option<(String, Option)>, + unpublished: bool, + stopped: bool, +} + +impl ClientState { + pub(super) fn capture_window_geometry( + &self, + window_id: &str, + read_native: impl FnOnce() -> WindowGeometry, + ) -> bool { + // A native getter can synchronously wait for the UI event loop. Do not + // hold ANY client-state lock until it returns, including during shutdown. + let geometry = read_native(); + let zoom = self + .zoom_levels + .lock() + .ok() + .and_then(|levels| levels.get(window_id).copied()) + .unwrap_or(DEFAULT_ZOOM_LEVEL); + self.queue_window_capture( + window_id, + geometry.bounds, + geometry.maximized, + geometry.fullscreen, + zoom, + ) + } + + // Neither lock below is held across disk I/O. One latest capture per known + // window bounds memory; unknown/removed/disabled windows cannot grow the queue. + pub(super) fn queue_window_capture( + &self, + window_id: &str, + bounds: Option, + maximized: bool, + fullscreen: bool, + zoom_factor: f64, + ) -> bool { + let Ok(state) = self.state.lock() else { + return false; + }; + if state.unsupported_future_envelope { + return false; + } + let Ok(mut pending) = self.pending_windows.lock() else { + return false; + }; + if pending.stopped { + return false; + } + let previous_window = if let Some((_, window)) = + pending.mutation.as_ref().filter(|(id, _)| id == window_id) + { + window.as_ref() + } else { + let Ok(record) = state.record(window_id) else { + return false; + }; + if !record.writes_enabled { + return false; + } + record.window.as_ref() + }; + let bounds = bounds.or_else(|| { + pending + .latest + .get(window_id) + .or(previous_window) + .map(|window| window.bounds.clone()) + }); + let Some(bounds) = bounds else { return false }; + pending.latest.insert( + window_id.to_string(), + NativeWindowState { + bounds, + maximized, + fullscreen, + zoom_factor, + }, + ); + true + } + + // Caller holds write_lock and state, before publication or record policy + // changes. Merging before rollback snapshots preserves pending captures even + // when a subsequent clear/disable/removal cannot be published. + pub(super) fn apply_window_captures( + &self, + state: &mut PersistedClientState, + ) -> Result<(), String> { + let mut pending = self + .pending_windows + .lock() + .map_err(|error| error.to_string())?; + if state.unsupported_future_envelope { + pending.latest.clear(); + return Ok(()); + } + let mut applied = false; + for (id, capture) in pending.latest.drain() { + if let Ok(record) = state.record_mut(&id) { + if record.writes_enabled { + record.window = Some(capture); + applied = true; + } + } + } + pending.unpublished |= applied; + Ok(()) + } + + pub(super) fn window_captures_published(&self) { + self.pending_windows + .lock() + .unwrap_or_else(|error| error.into_inner()) + .unpublished = false; + } + + // Caller holds state and write_lock. Only one record mutation can publish at + // once; other windows continue using their own policy and independent queue. + pub(super) fn preserve_window_capture_policy(&self, previous: &PersistedClientState, id: &str) { + let mut pending = self + .pending_windows + .lock() + .unwrap_or_else(|error| error.into_inner()); + pending.mutation = if previous.unsupported_future_envelope { + None + } else { + previous + .record(id) + .ok() + .filter(|record| record.writes_enabled) + .map(|record| (id.to_string(), record.window.clone())) + }; + } + + pub(super) fn finish_window_capture_policy(&self, state: &PersistedClientState, id: &str) { + let mut pending = self + .pending_windows + .lock() + .unwrap_or_else(|error| error.into_inner()); + pending.mutation = None; + // Successful destructive changes discard speculative events. On rollback, + // the restored record admits them and the next flush publishes the latest. + if state.unsupported_future_envelope + || !state.record(id).is_ok_and(|record| record.writes_enabled) + { + pending.latest.remove(id); + } + } + + pub(super) fn stop_window_captures(&self) { + self.pending_windows + .lock() + .unwrap_or_else(|error| error.into_inner()) + .stopped = true; + } + + // Caller holds write_lock, after the worker has joined and before ownership + // release. Admission is closed, so this is also safe if a wakeup was omitted. + pub(super) fn flush_pending_window_captures(&self) -> Result<(), String> { + if !self.is_primary() { + return Ok(()); + } + { + let mut state = self.state.lock().map_err(|error| error.to_string())?; + let pending = self + .pending_windows + .lock() + .map_err(|error| error.to_string())?; + if state.unsupported_future_envelope + || (pending.latest.is_empty() && !pending.unpublished) + { + return Ok(()); + } + drop(pending); + self.apply_window_captures(&mut state)?; + } + self.write_current_state() + } +} + +#[cfg(test)] +#[path = "window_updates_tests.rs"] +mod tests; diff --git a/packages/tauri-app/src-tauri/src/client_state/window_updates_tests.rs b/packages/tauri-app/src-tauri/src/client_state/window_updates_tests.rs new file mode 100644 index 000000000..5eaa52b5f --- /dev/null +++ b/packages/tauri-app/src-tauri/src/client_state/window_updates_tests.rs @@ -0,0 +1,545 @@ +use super::*; +use serde_json::{json, Value}; +use std::{ + fs, + sync::{ + atomic::{AtomicBool, Ordering}, + mpsc, Arc, Mutex, + }, + thread, + time::Duration, +}; + +const TIMEOUT: Duration = Duration::from_secs(5); + +fn bounds(x: i32) -> WindowBounds { + WindowBounds { + x, + y: 20, + width: 1200, + height: 800, + } +} + +fn geometry(x: i32) -> WindowGeometry { + WindowGeometry { + bounds: Some(bounds(x)), + maximized: false, + fullscreen: false, + } +} + +fn capture(state: &ClientState, id: &str, x: i32) -> bool { + state.capture_window_geometry(id, || geometry(x)) +} + +fn read_state(state: &ClientState) -> Value { + serde_json::from_slice(&fs::read(&state.state_path).unwrap()).unwrap() +} + +#[test] +fn captures_do_not_wait_for_disk_and_latest_geometry_is_flushed() { + let directory = tempfile::tempdir().unwrap(); + let block = Arc::new(AtomicBool::new(false)); + let writer_block = Arc::clone(&block); + let (entered, entries) = mpsc::channel(); + let (release, releases) = mpsc::channel(); + let releases = Mutex::new(releases); + let state = Arc::new( + ClientState::initialize_at_with_writer( + directory.path(), + Arc::new(move |path, bytes, valid| { + if writer_block.swap(false, Ordering::SeqCst) { + entered.send(()).unwrap(); + releases.lock().unwrap().recv_timeout(TIMEOUT * 2).unwrap(); + } + crate::client_state::write_atomically(path, bytes, valid) + }), + ) + .unwrap(), + ); + let id = state.active_window_id().unwrap(); + assert!(capture(&state, &id, 1)); + block.store(true, Ordering::SeqCst); + let writing = Arc::clone(&state); + let writer = thread::spawn(move || writing.flush().unwrap()); + entries.recv_timeout(TIMEOUT).unwrap(); + let capturing = Arc::clone(&state); + let window_id = id.clone(); + let (done, completion) = mpsc::channel(); + let events = thread::spawn(move || { + for x in 2..=1_000 { + assert!(capture(&capturing, &window_id, x)); + } + done.send(()).unwrap(); + }); + // A stalled fsync is released only AFTER move/resize handlers finish. + let captured = completion.recv_timeout(TIMEOUT); + release.send(()).unwrap(); + writer.join().unwrap(); + events.join().unwrap(); + captured.unwrap(); + assert_eq!(state.pending_windows.lock().unwrap().latest.len(), 1); + assert_eq!( + read_state(&state)["windows"][&id]["window"]["bounds"]["x"], + 1 + ); + state.flush().unwrap(); + assert_eq!( + read_state(&state)["windows"][&id]["window"]["bounds"]["x"], + 1_000 + ); +} + +#[test] +fn native_getters_run_without_any_client_state_lock() { + let directory = tempfile::tempdir().unwrap(); + let state = ClientState::initialize_at(directory.path()).unwrap(); + let id = state.active_window_id().unwrap(); + assert!(state.capture_window_geometry(&id, || { + assert!( + state.write_lock.try_lock().is_ok(), + "native getter holds disk lock" + ); + assert!( + state.state.try_lock().is_ok(), + "native getter holds state lock" + ); + assert!( + state.zoom_levels.try_lock().is_ok(), + "native getter holds zoom lock" + ); + assert!( + state.pending_windows.try_lock().is_ok(), + "native getter holds mailbox lock" + ); + geometry(10) + })); +} + +#[test] +fn background_native_capture_cannot_deadlock_main_thread_move_event() { + let directory = tempfile::tempdir().unwrap(); + let state = Arc::new(ClientState::initialize_at(directory.path()).unwrap()); + let id = state.active_window_id().unwrap(); + let background_state = Arc::clone(&state); + let background_id = id.clone(); + let (requested, requests) = mpsc::channel(); + let (release, releases) = mpsc::channel(); + let background = thread::spawn(move || { + background_state.capture_window_geometry(&background_id, || { + // Wry getters dispatch to the UI event loop then wait for its reply. + requested.send(()).unwrap(); + releases.recv_timeout(TIMEOUT * 2).unwrap(); + geometry(30) + }) + }); + requests.recv_timeout(TIMEOUT).unwrap(); + let ui_state = Arc::clone(&state); + let ui_id = id.clone(); + let (done, completion) = mpsc::channel(); + let ui = thread::spawn(move || { + assert!(capture(&ui_state, &ui_id, 20)); + done.send(()).unwrap(); + }); + let ui_free = completion.recv_timeout(TIMEOUT); + // Break a regressed cycle from the controller so the test never hangs forever. + release.send(()).unwrap(); + assert!(background.join().unwrap()); + ui.join().unwrap(); + ui_free.unwrap(); + state.flush().unwrap(); + assert_eq!( + read_state(&state)["windows"][&id]["window"]["bounds"], + json!(bounds(30)) + ); +} + +#[test] +fn maximization_preserves_latest_normal_bounds_and_windows_stay_independent() { + let directory = tempfile::tempdir().unwrap(); + let state = ClientState::initialize_at(directory.path()).unwrap(); + let first = state.active_window_id().unwrap(); + let second = uuid::Uuid::new_v4().to_string(); + state.add_window(second.clone()).unwrap(); + state + .save_snapshot_guarded_for(&first, json!({ "draft": "keep" }), || true) + .unwrap(); + assert!(capture(&state, &first, 30)); + assert!(capture(&state, &first, 90)); + assert!(state.queue_window_capture(&first, None, true, false, 1.5)); + assert!(capture(&state, &second, 60)); + for _ in 0..1_000 { + assert!(!capture(&state, &uuid::Uuid::new_v4().to_string(), 0)); + } + assert_eq!(state.pending_windows.lock().unwrap().latest.len(), 2); + state.flush().unwrap(); + let saved = read_state(&state); + assert_eq!( + saved["windows"][&first]["snapshot"], + json!({ "draft": "keep" }) + ); + assert_eq!( + saved["windows"][&first]["window"], + json!({ + "bounds": bounds(90), "maximized": true, "fullscreen": false, "zoomFactor": 1.5 + }) + ); + assert_eq!( + saved["windows"][&second]["window"]["bounds"], + json!(bounds(60)) + ); + // Fullscreen, minimized or unavailable normal bounds retain the last geometry. + assert!(state.queue_window_capture(&first, None, false, true, 1.5)); + state.flush().unwrap(); + assert_eq!( + read_state(&state)["windows"][&first]["window"]["bounds"], + json!(bounds(90)) + ); +} + +#[test] +fn clear_disable_and_removal_invalidate_pending_geometry() { + let directory = tempfile::tempdir().unwrap(); + let state = ClientState::initialize_at(directory.path()).unwrap(); + let id = state.active_window_id().unwrap(); + for clear in [false, true] { + assert!(capture(&state, &id, 70)); + if clear { + state.clear().unwrap(); + } else { + state.set_restore_enabled(false).unwrap(); + } + assert!(!capture(&state, &id, 80)); + state.set_restore_enabled(true).unwrap(); + state.flush().unwrap(); + assert!(read_state(&state)["windows"][&id].get("window").is_none()); + } + assert!(capture(&state, &id, 90)); + state.remove_window(&id).unwrap(); + assert!(!capture(&state, &id, 100)); + state.add_window(id.clone()).unwrap(); + state.flush().unwrap(); + assert!(read_state(&state)["windows"][&id].get("window").is_none()); +} + +#[test] +fn failed_writes_keep_captures_retryable_without_overwriting_newer_events() { + let directory = tempfile::tempdir().unwrap(); + let fail = Arc::new(AtomicBool::new(false)); + let writer_fail = Arc::clone(&fail); + let state = ClientState::initialize_at_with_writer( + directory.path(), + Arc::new(move |path, bytes, valid| { + if writer_fail.load(Ordering::SeqCst) { + return Err("simulated fsync failure".into()); + } + crate::client_state::write_atomically(path, bytes, valid) + }), + ) + .unwrap(); + let id = state.active_window_id().unwrap(); + assert!(capture(&state, &id, 20)); + state.flush().unwrap(); + fail.store(true, Ordering::SeqCst); + assert!(capture(&state, &id, 40)); + assert!(state.flush().is_err()); + assert_eq!( + read_state(&state)["windows"][&id]["window"]["bounds"], + json!(bounds(20)) + ); + fail.store(false, Ordering::SeqCst); + state.flush().unwrap(); + assert_eq!( + read_state(&state)["windows"][&id]["window"]["bounds"], + json!(bounds(40)) + ); + fail.store(true, Ordering::SeqCst); + assert!(capture(&state, &id, 60)); + assert!(state.flush().is_err()); + assert!(capture(&state, &id, 80)); + fail.store(false, Ordering::SeqCst); + state.flush().unwrap(); + assert_eq!( + read_state(&state)["windows"][&id]["window"]["bounds"], + json!(bounds(80)) + ); +} + +#[test] +fn failed_record_mutations_roll_back_queued_geometry_too() { + let directory = tempfile::tempdir().unwrap(); + let fail = Arc::new(AtomicBool::new(false)); + let writer_fail = Arc::clone(&fail); + let state = ClientState::initialize_at_with_writer( + directory.path(), + Arc::new(move |path, bytes, valid| { + if writer_fail.load(Ordering::SeqCst) { + return Err("simulated publication failure".into()); + } + crate::client_state::write_atomically(path, bytes, valid) + }), + ) + .unwrap(); + let id = state.active_window_id().unwrap(); + for operation in 0..3 { + assert!(capture(&state, &id, 10 + operation)); + fail.store(true, Ordering::SeqCst); + let result = match operation { + 0 => state.clear(), + 1 => state.set_restore_enabled(false), + _ => state.remove_window(&id), + }; + assert!(result.is_err()); + fail.store(false, Ordering::SeqCst); + state.flush().unwrap(); + assert_eq!( + read_state(&state)["windows"][&id]["window"]["bounds"], + json!(bounds(10 + operation)) + ); + } +} + +#[test] +fn release_drains_even_a_capture_whose_wakeup_has_not_been_sent() { + let directory = tempfile::tempdir().unwrap(); + let state = ClientState::initialize_at(directory.path()).unwrap(); + let id = state.active_window_id().unwrap(); + assert!(capture(&state, &id, 120)); + state.release_locks(); + assert_eq!( + read_state(&state)["windows"][&id]["window"]["bounds"], + json!(bounds(120)) + ); + assert!(!capture(&state, &id, 130)); + state + .window_flush + .schedule(|| panic!("worker restarted after release")) + .unwrap(); + state.release_locks(); + let successor = ClientState::initialize_at(directory.path()).unwrap(); + assert!(successor.is_primary()); + assert_eq!( + read_state(&successor)["windows"][&id]["window"]["bounds"], + json!(bounds(120)) + ); +} + +#[test] +fn release_retries_a_failed_flush_even_after_captures_were_merged() { + let directory = tempfile::tempdir().unwrap(); + let fail = Arc::new(AtomicBool::new(false)); + let writer_fail = Arc::clone(&fail); + let state = ClientState::initialize_at_with_writer( + directory.path(), + Arc::new(move |path, bytes, valid| { + if writer_fail.swap(false, Ordering::SeqCst) { + return Err("transient publication failure".into()); + } + crate::client_state::write_atomically(path, bytes, valid) + }), + ) + .unwrap(); + let id = state.active_window_id().unwrap(); + assert!(capture(&state, &id, 42)); + fail.store(true, Ordering::SeqCst); + assert!(state.flush().is_err()); + assert!(state.pending_windows.lock().unwrap().latest.is_empty()); + state.release_locks(); + assert_eq!( + read_state(&state)["windows"][&id]["window"]["bounds"], + json!(bounds(42)) + ); +} + +#[test] +fn scheduled_flush_drains_trailing_events_before_ownership_handoff() { + let directory = tempfile::tempdir().unwrap(); + let block = Arc::new(AtomicBool::new(false)); + let writer_block = Arc::clone(&block); + let (entered, entries) = mpsc::channel(); + let (release, releases) = mpsc::channel(); + let releases = Mutex::new(releases); + let state = Arc::new( + ClientState::initialize_at_with_writer( + directory.path(), + Arc::new(move |path, bytes, valid| { + if writer_block.swap(false, Ordering::SeqCst) { + entered.send(()).unwrap(); + releases.lock().unwrap().recv_timeout(TIMEOUT * 2).unwrap(); + } + crate::client_state::write_atomically(path, bytes, valid) + }), + ) + .unwrap(), + ); + let id = state.active_window_id().unwrap(); + assert!(capture(&state, &id, 10)); + block.store(true, Ordering::SeqCst); + let writing = Arc::clone(&state); + state + .window_flush + .schedule(move || writing.flush().unwrap()) + .unwrap(); + entries.recv_timeout(TIMEOUT).unwrap(); + assert!(capture(&state, &id, 20)); + let writing = Arc::clone(&state); + state + .window_flush + .schedule(move || writing.flush().unwrap()) + .unwrap(); + release.send(()).unwrap(); + state.release_locks(); + assert!(!capture(&state, &id, 30)); + let successor = ClientState::initialize_at(directory.path()).unwrap(); + assert!(successor.is_primary()); + assert_eq!( + read_state(&successor)["windows"][&id]["window"]["bounds"], + json!(bounds(20)) + ); +} + +#[test] +fn secondary_owner_and_future_envelope_cannot_publish_captures() { + let directory = tempfile::tempdir().unwrap(); + let primary = ClientState::initialize_at(directory.path()).unwrap(); + primary.flush().unwrap(); + let secondary = ClientState::initialize_at(directory.path()).unwrap(); + let before = fs::read(&primary.state_path).unwrap(); + capture(&secondary, &secondary.active_window_id().unwrap(), 50); + secondary.flush().unwrap(); + assert_eq!(fs::read(&primary.state_path).unwrap(), before); + let id = primary.active_window_id().unwrap(); + primary.state.lock().unwrap().unsupported_future_envelope = true; + assert!(!capture(&primary, &id, 60)); + primary.flush().unwrap(); + assert_eq!(fs::read(&primary.state_path).unwrap(), before); +} + +#[test] +fn failed_publication_preserves_concurrent_captures_for_mutating_and_other_windows() { + let directory = tempfile::tempdir().unwrap(); + let block = Arc::new(AtomicBool::new(false)); + let writer_block = Arc::clone(&block); + let (entered, entries) = mpsc::channel(); + let (release, releases) = mpsc::channel(); + let releases = Mutex::new(releases); + let state = Arc::new( + ClientState::initialize_at_with_writer( + directory.path(), + Arc::new(move |path, bytes, valid| { + if writer_block.swap(false, Ordering::SeqCst) { + entered.send(()).unwrap(); + releases.lock().unwrap().recv_timeout(TIMEOUT * 2).unwrap(); + return Err("simulated publication failure".into()); + } + crate::client_state::write_atomically(path, bytes, valid) + }), + ) + .unwrap(), + ); + let first = state.active_window_id().unwrap(); + let second = uuid::Uuid::new_v4().to_string(); + state.add_window(second.clone()).unwrap(); + state + .zoom_levels + .lock() + .unwrap() + .insert(second.clone(), 1.75); + for operation in 0..4 { + assert!(capture(&state, &first, 10 + operation)); + assert!(capture(&state, &second, 20)); + state.flush().unwrap(); + let writing = Arc::clone(&state); + let id = first.clone(); + block.store(true, Ordering::SeqCst); + let writer = thread::spawn(move || match operation { + 0 => writing.save_snapshot_guarded_for(&id, json!({"draft":"new"}), || true), + 1 => writing.clear_guarded(&id, || true), + 2 => writing.set_restore_enabled_guarded(&id, false, || true), + _ => writing.remove_window(&id), + }); + entries.recv_timeout(TIMEOUT).unwrap(); + let capturing = Arc::clone(&state); + let id = second.clone(); + let changing_id = first.clone(); + let (done, completion) = mpsc::channel(); + let events = thread::spawn(move || { + assert!(capture(&capturing, &id, 100 + operation)); + assert!(capture(&capturing, &changing_id, 200 + operation)); + done.send(()).unwrap(); + }); + let captured = completion.recv_timeout(TIMEOUT); + release.send(()).unwrap(); + assert!(writer.join().unwrap().is_err()); + events.join().unwrap(); + captured.unwrap(); + state.flush().unwrap(); + let saved = read_state(&state); + assert_eq!( + saved["windows"][&first]["window"]["bounds"], + json!(bounds(200 + operation)) + ); + assert_eq!( + saved["windows"][&second]["window"]["bounds"], + json!(bounds(100 + operation)) + ); + assert_eq!(saved["windows"][&second]["window"]["zoomFactor"], 1.75); + } +} + +#[test] +fn successful_destructive_publication_discards_speculative_captures() { + let directory = tempfile::tempdir().unwrap(); + let during_write = Arc::new(Mutex::new(None::>)); + let callback = Arc::clone(&during_write); + let state = Arc::new( + ClientState::initialize_at_with_writer( + directory.path(), + Arc::new(move |path, bytes, valid| { + // The production capture doesn't acquire write_lock. Assert it can run + // inside the publication adapter while the old admission policy is saved. + if let Some(state) = callback + .lock() + .unwrap() + .take() + .and_then(|state| state.upgrade()) + { + let pending = state.pending_windows.lock().unwrap(); + let id = pending.mutation.as_ref().unwrap().0.clone(); + drop(pending); + let (done, completion) = mpsc::channel(); + let capturing = thread::spawn(move || { + let _ = done.send(capture(&state, &id, 99)); + }); + // On a regression, return and release write_lock rather than leaving + // the test (or its capture thread) hung behind its own publication. + let captured = completion + .recv_timeout(TIMEOUT) + .map_err(|error| error.to_string())?; + capturing.join().unwrap(); + assert!(captured); + } + crate::client_state::write_atomically(path, bytes, valid) + }), + ) + .unwrap(), + ); + let id = state.active_window_id().unwrap(); + for operation in 0..3 { + assert!(capture(&state, &id, 10)); + *during_write.lock().unwrap() = Some(Arc::downgrade(&state)); + match operation { + 0 => state.clear().unwrap(), + 1 => state.set_restore_enabled(false).unwrap(), + _ => state.remove_window(&id).unwrap(), + }; + assert!(state.pending_windows.lock().unwrap().latest.is_empty()); + if operation == 2 { + state.add_window(id.clone()).unwrap(); + } + state.set_restore_enabled(true).unwrap(); + state.flush().unwrap(); + assert!(read_state(&state)["windows"][&id].get("window").is_none()); + } +}