fix: don't discard commit-group result slot before it is populated

The .take() in lock() unconditionally discarded the Arc before checking
whether the OnceLock had been populated.  If lock() fired between
finalize_transaction (which stages the intent) and mv_commit_group_commit
(which populates the slot), the slot was lost and the connection would
silently fall back to the interval path.

Switch to .as_ref() so the slot survives unpopulated reads, and only
clear it (via direct assignment) after successfully reading the result.

Also adds unit tests for the CommitGroupResultSlot and bumps version
to 0.3.7.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: losfair

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "mvclient"
-version = "0.3.6"
+version = "0.3.7"
 edition = "2021"
 license = "Apache-2.0"
 authors = ["Heyang Zhou <heyang.zhou@icloud.com>"]

mvclient/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mvfs"
-version = "0.3.6"
+version = "0.3.7"
 edition = "2021"
 license = "Apache-2.0"
 authors = ["Heyang Zhou <heyang.zhou@icloud.com>"]
@@ -15,7 +15,7 @@ anyhow = "1"
 thiserror = "2"
 futures = "0.3"
 tokio = { version = "1", features = ["full"] }
-mvclient = { path = "../mvclient", version = "0.3.6" }
+mvclient = { path = "../mvclient", version = "0.3.7" }
 tracing = "0.1"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

mvfs/Cargo.toml

@@ -3,7 +3,7 @@ use std::{
     collections::HashMap,
     future::Future,
     pin::Pin,
-    sync::Arc,
+    sync::{Arc, OnceLock},
 };
 
 use anyhow::{bail, Context, Result};
@@ -15,6 +15,7 @@ pub struct CommitGroup {
     intents: Vec<NamespaceCommitIntent>,
     client: Option<Arc<MultiVersionClient>>,
     dp: Option<Url>,
+    result_slot: Arc<CommitGroupResultSlot>,
 }
 
 impl Default for CommitGroup {
@@ -23,6 +24,9 @@ impl Default for CommitGroup {
             intents: Vec::new(),
             client: None,
             dp: None,
+            result_slot: Arc::new(CommitGroupResultSlot {
+                inner: OnceLock::new(),
+            }),
         }
     }
 }
@@ -38,22 +42,34 @@ pub enum CommitOutput {
     Conflict,
 }
 
-/// Stores the result of the last successful commit group so that participating
-/// connections can update their page caches and `last_known_write_version` when
-/// they next acquire a lock.
-struct CommitGroupResult {
+/// Shared slot that `commit()` writes into and each participating
+/// `Connection` reads from on its next `lock()` call.  Thread-safe
+/// because connections may migrate threads after the commit group ends.
+pub struct CommitGroupResultSlot {
+    inner: OnceLock<CommitGroupResultData>,
+}
+
+struct CommitGroupResultData {
     version: String,
-    /// Namespace keys that staged intents in this commit group.
-    ns_keys: Vec<String>,
     /// Per-namespace changelog of pages modified by *concurrent* transactions
     /// between each namespace's assumed version and the committed version.
     /// A missing entry means the interval was too large to compute.
     changelog: HashMap<String, Vec<u32>>,
 }
 
+impl CommitGroupResultSlot {
+    /// Returns the committed version and the changelog entry for `ns_key`.
+    /// `changelog` is `None` when the server could not compute the interval;
+    /// the caller should do a full cache invalidation in that case.
+    pub fn get_for_ns(&self, ns_key: &str) -> Option<(String, Option<Vec<u32>>)> {
+        self.inner.get().map(|data| {
+            (data.version.clone(), data.changelog.get(ns_key).cloned())
+        })
+    }
+}
+
 thread_local! {
     static CURRENT_COMMIT_GROUP: RefCell<Option<CommitGroup>> = RefCell::new(None);
-    static LAST_COMMIT_GROUP_RESULT: RefCell<Option<CommitGroupResult>> = RefCell::new(None);
 }
 
 pub fn begin() -> Result<()> {
@@ -64,11 +80,7 @@ pub fn begin() -> Result<()> {
         }
         *cg = Some(CommitGroup::default());
         Ok(())
-    })?;
-    LAST_COMMIT_GROUP_RESULT.with(|r| {
-        *r.borrow_mut() = None;
-    });
-    Ok(())
+    })
 }
 
 pub fn is_active() -> bool {
@@ -90,7 +102,7 @@ pub fn append_intent(
     client: &Arc<MultiVersionClient>,
     dp: Option<&Url>,
     intent: NamespaceCommitIntent,
-) -> Result<()> {
+) -> Result<Arc<CommitGroupResultSlot>> {
     CURRENT_COMMIT_GROUP.with(|cg| {
         let mut cg = cg.borrow_mut();
         let cg = cg
@@ -104,8 +116,9 @@ pub fn append_intent(
             cg.dp = dp.cloned();
         }
 
+        let slot = cg.result_slot.clone();
         cg.intents.push(intent);
-        Ok(())
+        Ok(slot)
     })
 }
 
@@ -121,7 +134,7 @@ pub fn commit(
         return Ok(CommitOutput::Empty);
     }
 
-    let ns_keys: Vec<String> = cg.intents.iter().map(|i| i.init.ns_key.clone()).collect();
+    let result_slot = cg.result_slot.clone();
 
     let client = cg
         .client
@@ -133,12 +146,9 @@ pub fn commit(
 
     Ok(match result {
         Some(result) => {
-            LAST_COMMIT_GROUP_RESULT.with(|r| {
-                *r.borrow_mut() = Some(CommitGroupResult {
-                    version: result.version.clone(),
-                    ns_keys,
-                    changelog: result.changelog.clone(),
-                });
+            let _ = result_slot.inner.set(CommitGroupResultData {
+                version: result.version.clone(),
+                changelog: result.changelog.clone(),
             });
             CommitOutput::Committed(result)
         }
@@ -152,24 +162,6 @@ pub fn rollback() -> Result<()> {
     Ok(())
 }
 
-/// If a commit group recently succeeded and `ns_key` participated in it,
-/// returns `(committed_version, changelog)`.  `changelog` is `None` when
-/// the server could not compute the interval (too many concurrent changes);
-/// in that case the caller should do a full cache invalidation.
-pub fn take_commit_group_result_for_ns(ns_key: &str) -> Option<(String, Option<Vec<u32>>)> {
-    LAST_COMMIT_GROUP_RESULT.with(|r| {
-        let r = r.borrow();
-        r.as_ref().and_then(|result| {
-            if result.ns_keys.iter().any(|k| k == ns_key) {
-                let changelog = result.changelog.get(ns_key).cloned();
-                Some((result.version.clone(), changelog))
-            } else {
-                None
-            }
-        })
-    })
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -233,4 +225,47 @@ mod tests {
 
         rollback().unwrap();
     }
+
+    #[test]
+    fn result_slot_empty_before_commit() {
+        let slot = Arc::new(CommitGroupResultSlot {
+            inner: OnceLock::new(),
+        });
+        assert!(slot.get_for_ns("test").is_none());
+    }
+
+    #[test]
+    fn result_slot_returns_changelog_after_population() {
+        let slot = Arc::new(CommitGroupResultSlot {
+            inner: OnceLock::new(),
+        });
+
+        let _ = slot.inner.set(CommitGroupResultData {
+            version: "abc123".into(),
+            changelog: HashMap::from([("ns1".into(), vec![1, 2, 3])]),
+        });
+
+        // Participating namespace gets version + changelog
+        let (version, changelog) = slot.get_for_ns("ns1").unwrap();
+        assert_eq!(version, "abc123");
+        assert_eq!(changelog, Some(vec![1, 2, 3]));
+
+        // Non-participating namespace gets version but no changelog entry
+        let (version, changelog) = slot.get_for_ns("ns2").unwrap();
+        assert_eq!(version, "abc123");
+        assert!(changelog.is_none());
+    }
+
+    #[test]
+    fn append_intent_returns_shared_slot() {
+        begin().unwrap();
+
+        let slot1 = append_intent(&test_client(), None, test_intent()).unwrap();
+        let slot2 = append_intent(&test_client(), None, test_intent()).unwrap();
+
+        // Both intents share the same slot
+        assert!(Arc::ptr_eq(&slot1, &slot2));
+
+        rollback().unwrap();
+    }
 }

mvfs/src/commit_group.rs

@@ -137,6 +137,7 @@ impl MultiVersionVfs {
             write_buffer: HashMap::new(),
             virtual_version_counter: 0,
             last_known_write_version: None,
+            pending_commit_group_result: None,
         };
         Ok(conn)
     }
@@ -163,6 +164,11 @@ pub struct Connection {
     virtual_version_counter: u32,
 
     last_known_write_version: Option<String>,
+
+    /// Shared slot populated by `commit_group::commit()` after a successful
+    /// commit group.  Consumed on the next `lock()` call to apply the
+    /// changelog and advance `last_known_write_version`.
+    pending_commit_group_result: Option<Arc<commit_group::CommitGroupResultSlot>>,
 }
 
 #[derive(Default)]
@@ -448,8 +454,10 @@ impl Connection {
             }
 
             if let Some(intent) = intent {
-                commit_group::append_intent(&self.client, self.dp.as_ref(), intent)
-                    .expect("failed to append to commit group");
+                let result_slot =
+                    commit_group::append_intent(&self.client, self.dp.as_ref(), intent)
+                        .expect("failed to append to commit group");
+                self.pending_commit_group_result = Some(result_slot);
                 tracing::info!("added intent to commit group");
             } else {
                 tracing::info!("transaction is empty (commit group)");
@@ -636,11 +644,14 @@ impl Connection {
             // participated, apply the changelog and advance
             // last_known_write_version so the subsequent interval request
             // covers only changes *after* the commit group.
-            if let Some((committed_version, changelog)) =
-                commit_group::take_commit_group_result_for_ns(
-                    self.client.config().ns_key.as_str(),
-                )
+            if let Some((committed_version, changelog)) = self
+                .pending_commit_group_result
+                .as_ref()
+                .and_then(|slot| {
+                    slot.get_for_ns(self.client.config().ns_key.as_str())
+                })
             {
+                self.pending_commit_group_result = None;
                 match changelog {
                     Some(pages) => {
                         if !pages.is_empty() {

mvfs/src/vfs.rs

@@ -1,6 +1,6 @@
 [package]
 name = "mvsqlite-fuse"
-version = "0.3.6"
+version = "0.3.7"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -11,7 +11,7 @@ thiserror = "2"
 tokio = { version = "1", features = ["full"] }
 log = "0.4"
 rand = "0.9"
-mvclient = { path = "../mvclient", version = "0.3.6" }
+mvclient = { path = "../mvclient", version = "0.3.7" }
 tracing = "0.1"
 tracing-subscriber = { version = "0.3.16", features = ["env-filter", "fmt", "tracing-log", "json"] }
 libc = "0.2"
@@ -20,7 +20,7 @@ serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 lazy_static = "1.4.0"
 reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] }
-mvfs = { path = "../mvfs", version = "0.3.6" }
+mvfs = { path = "../mvfs", version = "0.3.7" }
 fuser = "0.14"
 structopt = "0.3.26"
 indexmap = "2"

mvsqlite-fuse/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mvsqlite"
-version = "0.3.6"
+version = "0.3.7"
 edition = "2021"
 license = "Apache-2.0"
 authors = ["Heyang Zhou <heyang.zhou@icloud.com>"]
@@ -17,7 +17,7 @@ tokio = { version = "1", features = ["full"] }
 log = "0.4"
 rand = "0.9"
 stackful = "0.1.5"
-mvclient = { path = "../mvclient", version = "0.3.6" }
+mvclient = { path = "../mvclient", version = "0.3.7" }
 tracing = "0.1"
 ctor = "0.2"
 tracing-subscriber = { version = "0.3.16", features = ["env-filter", "fmt", "json"], optional = true }
@@ -27,7 +27,7 @@ serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 lazy_static = "1.4.0"
 reqwest = { version = "0.12", default-features = false, features = ["http2"] }
-mvfs = { path = "../mvfs", version = "0.3.6" }
+mvfs = { path = "../mvfs", version = "0.3.7" }
 
 [features]
 default = ["loadext", "syscall", "rustls-tls", "global-init"]

mvsqlite/Cargo.toml

@@ -1,12 +1,12 @@
 [package]
 name = "mvstore-stress"
-version = "0.3.6"
+version = "0.3.7"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-mvclient = { path = "../mvclient", version = "0.3.6" }
+mvclient = { path = "../mvclient", version = "0.3.7" }
 anyhow = "1"
 thiserror = "2"
 tokio = { version = "1", features = ["full"] }