Serialize playback speed updates

Signed-off-by: John Tennant <jtennant@squareup.com>

desktop/src-tauri/Cargo.lock

@@ -1561,7 +1561,7 @@ version = "3.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.48.0",
 ]
 
 [[package]]
@@ -1799,7 +1799,7 @@ dependencies = [
  "wasm-bindgen",
  "wasm-bindgen-futures",
  "web-sys",
- "windows 0.62.2",
+ "windows 0.61.3",
 ]
 
 [[package]]
@@ -2284,7 +2284,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090"
 dependencies = [
  "data-encoding",
- "syn 2.0.118",
+ "syn 1.0.109",
 ]
 
 [[package]]
@@ -3251,8 +3251,8 @@ dependencies = [
  "libc",
  "log",
  "rustversion",
- "windows-link 0.2.1",
- "windows-result 0.4.1",
+ "windows-link 0.1.3",
+ "windows-result 0.3.4",
 ]
 
 [[package]]
@@ -3997,7 +3997,7 @@ dependencies = [
  "tokio",
  "tower-service",
  "tracing",
- "windows-registry 0.6.1",
+ "windows-registry 0.5.3",
 ]
 
 [[package]]
@@ -4012,7 +4012,7 @@ dependencies = [
  "js-sys",
  "log",
  "wasm-bindgen",
- "windows-core 0.62.2",
+ "windows-core 0.61.2",
 ]
 
 [[package]]
@@ -6410,7 +6410,7 @@ version = "0.7.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "680998035259dcfcafe653688bf2aa6d3e2dc05e98be6ab46afb089dc84f1df8"
 dependencies = [
- "proc-macro-crate 3.5.0",
+ "proc-macro-crate 1.3.1",
  "proc-macro2",
  "quote",
  "syn 2.0.118",
@@ -6996,7 +6996,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7d8fae84b431384b68627d0f9b3b1245fcf9f46f6c0e3dc902e9dce64edd1967"
 dependencies = [
  "libc",
- "windows-sys 0.61.2",
+ "windows-sys 0.45.0",
 ]
 
 [[package]]
@@ -7755,7 +7755,7 @@ version = "0.14.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042"
 dependencies = [
- "heck 0.5.0",
+ "heck 0.4.1",
  "itertools",
  "log",
  "multimap",
@@ -10646,7 +10646,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
 dependencies = [
  "fastrand",
- "getrandom 0.4.3",
+ "getrandom 0.3.4",
  "once_cell",
  "rustix 1.1.4",
  "windows-sys 0.61.2",
@@ -12345,7 +12345,7 @@ version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.48.0",
 ]
 
 [[package]]
@@ -12974,8 +12974,8 @@ dependencies = [
  "log",
  "serde",
  "thiserror 2.0.18",
- "windows 0.62.2",
- "windows-core 0.62.2",
+ "windows 0.61.3",
+ "windows-core 0.61.2",
 ]
 
 [[package]]

desktop/src-tauri/src/huddle/playback_speed.rs

@@ -29,24 +29,38 @@ const UNITY_EPSILON: f32 = 0.000_1;
 
 /// Lock-free shared control read by the TTS worker before each synthesis chunk.
 #[derive(Clone, Debug)]
-pub struct PlaybackSpeedControl(Arc<AtomicU32>);
+pub struct PlaybackSpeedControl {
+    speed: Arc<AtomicU32>,
+    transition: Arc<tokio::sync::Mutex<()>>,
+}
 
 impl Default for PlaybackSpeedControl {
     fn default() -> Self {
-        Self(Arc::new(AtomicU32::new(DEFAULT_PLAYBACK_SPEED.to_bits())))
+        Self {
+            speed: Arc::new(AtomicU32::new(DEFAULT_PLAYBACK_SPEED.to_bits())),
+            transition: Arc::new(tokio::sync::Mutex::new(())),
+        }
     }
 }
 
 impl PlaybackSpeedControl {
     /// Return the current generated-speech playback speed.
     pub fn get(&self) -> f32 {
-        f32::from_bits(self.0.load(Ordering::Acquire))
+        f32::from_bits(self.speed.load(Ordering::Acquire))
     }
 
     /// Update the in-memory speed after validation.
     pub fn set(&self, speed: f32) -> Result<(), String> {
         validate_speed(speed)?;
-        self.0.store(speed.to_bits(), Ordering::Release);
+        self.speed.store(speed.to_bits(), Ordering::Release);
+        Ok(())
+    }
+
+    async fn persist_to_path(&self, path: &Path, speed: f32) -> Result<(), String> {
+        validate_speed(speed)?;
+        let _transition = self.transition.lock().await;
+        save_to_path(path, speed)?;
+        self.speed.store(speed.to_bits(), Ordering::Release);
         Ok(())
     }
 }
@@ -71,14 +85,15 @@ pub fn get_tts_playback_speed(state: State<'_, AppState>) -> f32 {
 
 /// Persist and apply the global generated-speech playback speed.
 #[tauri::command]
-pub fn set_tts_playback_speed(
+pub async fn set_tts_playback_speed(
     speed: f32,
     app: AppHandle,
     state: State<'_, AppState>,
 ) -> Result<(), String> {
-    validate_speed(speed)?;
-    save_to_path(&settings_path(&app)?, speed)?;
-    state.tts_playback_speed.set(speed)
+    state
+        .tts_playback_speed
+        .persist_to_path(&settings_path(&app)?, speed)
+        .await
 }
 
 fn settings_path(app: &AppHandle) -> Result<PathBuf, String> {
@@ -320,6 +335,44 @@ mod tests {
         assert!(load_from_path(&path).is_err());
     }
 
+    #[tokio::test]
+    async fn serialized_persistence_finishes_with_the_latest_speed() {
+        let directory = tempfile::tempdir().expect("tempdir");
+        let path = directory.path().join(SETTINGS_FILE);
+        let control = PlaybackSpeedControl::default();
+        let transition = control.transition.lock().await;
+
+        let (first_enqueued_tx, first_enqueued_rx) = tokio::sync::oneshot::channel();
+        let first_control = control.clone();
+        let first_path = path.clone();
+        let first = tokio::spawn(async move {
+            first_enqueued_tx.send(()).expect("signal first waiter");
+            first_control.persist_to_path(&first_path, 1.25).await
+        });
+        first_enqueued_rx.await.expect("first waiter started");
+        tokio::task::yield_now().await;
+
+        let (second_enqueued_tx, second_enqueued_rx) = tokio::sync::oneshot::channel();
+        let second_control = control.clone();
+        let second_path = path.clone();
+        let second = tokio::spawn(async move {
+            second_enqueued_tx.send(()).expect("signal second waiter");
+            second_control.persist_to_path(&second_path, 1.5).await
+        });
+        second_enqueued_rx.await.expect("second waiter started");
+        tokio::task::yield_now().await;
+
+        drop(transition);
+        first.await.expect("join first save").expect("first save");
+        second
+            .await
+            .expect("join second save")
+            .expect("second save");
+
+        assert_eq!(load_from_path(&path).expect("load final speed"), 1.5);
+        assert_eq!(control.get(), 1.5);
+    }
+
     fn zero_crossing_frequency(samples: &[f32], sample_rate: u32) -> f32 {
         let crossings = samples
             .windows(2)

desktop/src/features/settings/ui/SpeechPlaybackSettings.tsx

@@ -1,6 +1,7 @@
 import { useEffect, useRef, useState } from "react";
 import { getTtsPlaybackSpeed, setTtsPlaybackSpeed } from "@/shared/api/tauri";
 import { Button } from "@/shared/ui/button";
+import { commitPlaybackSpeed } from "./playbackSpeedPersistence";
 import { SettingsOptionGroup, SettingsOptionRow } from "./SettingsOptionGroup";
 
 const DEFAULT_SPEED = 1;
@@ -11,14 +12,17 @@ const SPEED_STEP = 0.05;
 export function SpeechPlaybackSettings() {
   const [speed, setSpeed] = useState(DEFAULT_SPEED);
   const [error, setError] = useState<string | null>(null);
-  const committedSpeed = useRef(DEFAULT_SPEED);
+  const confirmedSpeed = useRef(DEFAULT_SPEED);
+  const desiredSpeed = useRef(DEFAULT_SPEED);
+  const flushPromise = useRef<Promise<void> | null>(null);
 
   useEffect(() => {
     let active = true;
     getTtsPlaybackSpeed()
       .then((savedSpeed) => {
         if (active) {
-          committedSpeed.current = savedSpeed;
+          confirmedSpeed.current = savedSpeed;
+          desiredSpeed.current = savedSpeed;
           setSpeed(savedSpeed);
         }
       })
@@ -30,21 +34,12 @@ export function SpeechPlaybackSettings() {
     };
   }, []);
 
-  const commitSpeed = async (nextSpeed: number) => {
-    const previous = committedSpeed.current;
-    if (nextSpeed === previous) return;
-    committedSpeed.current = nextSpeed;
-    setSpeed(nextSpeed);
-    setError(null);
-    try {
-      await setTtsPlaybackSpeed(nextSpeed);
-    } catch (cause) {
-      if (committedSpeed.current === nextSpeed) {
-        committedSpeed.current = previous;
-        setSpeed(previous);
-        setError(String(cause));
-      }
-    }
+  const commitSpeed = (nextSpeed: number) => {
+    commitPlaybackSpeed(
+      { confirmedSpeed, desiredSpeed, flushPromise },
+      { persist: setTtsPlaybackSpeed, setError, setSpeed },
+      nextSpeed,
+    );
   };
 
   return (

desktop/src/features/settings/ui/playbackSpeedPersistence.test.mjs

@@ -0,0 +1,88 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import { commitPlaybackSpeed } from "./playbackSpeedPersistence.ts";
+
+function deferred() {
+  let resolve;
+  let reject;
+  const promise = new Promise((resolvePromise, rejectPromise) => {
+    resolve = resolvePromise;
+    reject = rejectPromise;
+  });
+  return { promise, reject, resolve };
+}
+
+function harness() {
+  const calls = [];
+  const errors = [];
+  const saves = [];
+  const speeds = [];
+  const state = {
+    confirmedSpeed: { current: 1 },
+    desiredSpeed: { current: 1 },
+    flushPromise: { current: null },
+  };
+  const callbacks = {
+    persist: (speed) => {
+      calls.push(speed);
+      const save = deferred();
+      saves.push(save);
+      return save.promise;
+    },
+    setError: (error) => errors.push(error),
+    setSpeed: (speed) => speeds.push(speed),
+  };
+  return { callbacks, calls, errors, saves, speeds, state };
+}
+
+async function settle(state) {
+  while (state.flushPromise.current) {
+    await state.flushPromise.current;
+  }
+}
+
+test("rapid intents persist serially and finish at the latest speed", async () => {
+  const h = harness();
+  commitPlaybackSpeed(h.state, h.callbacks, 1.25);
+  commitPlaybackSpeed(h.state, h.callbacks, 1.5);
+  assert.deepEqual(h.calls, [1.25]);
+
+  h.saves[0].resolve();
+  await Promise.resolve();
+  assert.deepEqual(h.calls, [1.25, 1.5]);
+  h.saves[1].resolve();
+  await settle(h.state);
+
+  assert.equal(h.state.confirmedSpeed.current, 1.5);
+  assert.equal(h.state.desiredSpeed.current, 1.5);
+});
+
+test("a stale failure does not roll back a newer successful intent", async () => {
+  const h = harness();
+  commitPlaybackSpeed(h.state, h.callbacks, 1.25);
+  commitPlaybackSpeed(h.state, h.callbacks, 1.5);
+
+  h.saves[0].reject(new Error("first failed"));
+  await Promise.resolve();
+  assert.deepEqual(h.calls, [1.25, 1.5]);
+  h.saves[1].resolve();
+  await settle(h.state);
+
+  assert.equal(h.state.confirmedSpeed.current, 1.5);
+  assert.equal(h.state.desiredSpeed.current, 1.5);
+  assert.deepEqual(h.speeds, [1.25, 1.5]);
+  assert.deepEqual(h.errors, [null, null]);
+});
+
+test("the latest failure rolls back to the last confirmed speed", async () => {
+  const h = harness();
+  commitPlaybackSpeed(h.state, h.callbacks, 1.25);
+  h.saves[0].reject(new Error("save failed"));
+  await settle(h.state);
+
+  assert.equal(h.state.confirmedSpeed.current, 1);
+  assert.equal(h.state.desiredSpeed.current, 1);
+  assert.deepEqual(h.speeds, [1.25, 1]);
+  assert.equal(h.errors.at(-1), "Error: save failed");
+});

desktop/src/features/settings/ui/playbackSpeedPersistence.ts

@@ -0,0 +1,62 @@
+type MutableValue<T> = {
+  current: T;
+};
+
+export type PlaybackSpeedPersistenceState = {
+  confirmedSpeed: MutableValue<number>;
+  desiredSpeed: MutableValue<number>;
+  flushPromise: MutableValue<Promise<void> | null>;
+};
+
+type PlaybackSpeedPersistenceCallbacks = {
+  persist: (speed: number) => Promise<void>;
+  setSpeed: (speed: number) => void;
+  setError: (error: string | null) => void;
+};
+
+export function commitPlaybackSpeed(
+  state: PlaybackSpeedPersistenceState,
+  callbacks: PlaybackSpeedPersistenceCallbacks,
+  nextSpeed: number,
+) {
+  state.desiredSpeed.current = nextSpeed;
+  callbacks.setSpeed(nextSpeed);
+  callbacks.setError(null);
+  ensurePlaybackSpeedFlush(state, callbacks);
+}
+
+function ensurePlaybackSpeedFlush(
+  state: PlaybackSpeedPersistenceState,
+  callbacks: PlaybackSpeedPersistenceCallbacks,
+) {
+  if (state.flushPromise.current) return;
+
+  const flush = flushPlaybackSpeed(state, callbacks).finally(() => {
+    if (state.flushPromise.current !== flush) return;
+    state.flushPromise.current = null;
+    if (state.desiredSpeed.current !== state.confirmedSpeed.current) {
+      ensurePlaybackSpeedFlush(state, callbacks);
+    }
+  });
+  state.flushPromise.current = flush;
+}
+
+async function flushPlaybackSpeed(
+  state: PlaybackSpeedPersistenceState,
+  callbacks: PlaybackSpeedPersistenceCallbacks,
+) {
+  while (state.desiredSpeed.current !== state.confirmedSpeed.current) {
+    const target = state.desiredSpeed.current;
+    try {
+      await callbacks.persist(target);
+      state.confirmedSpeed.current = target;
+    } catch (cause) {
+      if (state.desiredSpeed.current === target) {
+        state.desiredSpeed.current = state.confirmedSpeed.current;
+        callbacks.setSpeed(state.confirmedSpeed.current);
+        callbacks.setError(String(cause));
+        return;
+      }
+    }
+  }
+}