diff --git a/Cargo.lock b/Cargo.lock
index 6928852..7bb5a46 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -6344,6 +6344,7 @@ dependencies = [
 name = "lance-context-core"
 version = "0.6.5"
 dependencies = [
+ "arc-swap",
  "arrow-array 58.3.0",
  "arrow-ipc 58.3.0",
  "arrow-json 58.3.0",
diff --git a/crates/lance-context-core/Cargo.toml b/crates/lance-context-core/Cargo.toml
index d77daf2..c38b516 100644
--- a/crates/lance-context-core/Cargo.toml
+++ b/crates/lance-context-core/Cargo.toml
@@ -18,6 +18,7 @@ default = ["metrics"]
 metrics = ["dep:metrics"]
 
 [dependencies]
+arc-swap = "1"
 base64 = "0.22"
 arrow-array = "58"
 arrow-ipc = "58"
diff --git a/crates/lance-context-core/src/datagen_store.rs b/crates/lance-context-core/src/datagen_store.rs
index b967a46..0909972 100644
--- a/crates/lance-context-core/src/datagen_store.rs
+++ b/crates/lance-context-core/src/datagen_store.rs
@@ -125,8 +125,8 @@ impl DatagenStore {
     }
 
     #[must_use]
-    pub fn uri(&self) -> &str {
-        self.base.dataset.uri()
+    pub fn uri(&self) -> String {
+        self.base.uri()
     }
 
     #[must_use]
@@ -141,7 +141,7 @@ impl DatagenStore {
     }
 
     /// Refresh this handle to the latest base-table manifest.
-    pub async fn refresh_latest(&mut self) -> LanceResult<()> {
+    pub async fn refresh_latest(&self) -> LanceResult<()> {
         self.base.refresh_latest().await
     }
 
@@ -150,7 +150,7 @@ impl DatagenStore {
     /// The supplied slice is persisted as one MemWAL generation. Callers should
     /// include FIELD_* events and the corresponding STEP_COMPLETED marker in
     /// the same call so a crash cannot expose a partially checkpointed step.
-    pub async fn append(&mut self, events: &[DatagenEvent]) -> LanceResult<u64> {
+    pub async fn append(&self, events: &[DatagenEvent]) -> LanceResult<u64> {
         if events.is_empty() {
             return Ok(self.base.version());
         }
@@ -189,7 +189,7 @@ impl DatagenStore {
     }
 
     /// Gracefully stop this store's resident MemWAL writer.
-    pub async fn close(&mut self) -> LanceResult<()> {
+    pub async fn close(&self) -> LanceResult<()> {
         self.base.close().await
     }
 
@@ -336,9 +336,11 @@ impl DatagenStore {
             }
         }
 
-        Ok(Self::get_blob_from_dataset(&self.base.dataset, event_id)
-            .await?
-            .flatten())
+        Ok(
+            Self::get_blob_from_dataset(self.base.current_dataset().as_ref(), event_id)
+                .await?
+                .flatten(),
+        )
     }
 
     /// Materialize a folded item's blob field by name, resolving the `event_id` for the caller.
@@ -365,7 +367,7 @@ impl DatagenStore {
 
     /// Merge every currently flushed generation owned by this writer into the
     /// base table.
-    pub async fn cleanup_own_shard(&mut self) -> LanceResult<usize> {
+    pub async fn cleanup_own_shard(&self) -> LanceResult<usize> {
         self.base.cleanup_own_shard().await
     }
 
@@ -380,7 +382,7 @@ impl DatagenStore {
     /// the shared base table and Lance treats two concurrent `Rewrite` commits
     /// as a conflict.
     pub async fn compact(
-        &mut self,
+        &self,
         options: Option<CompactionConfig>,
     ) -> LanceResult<CompactionMetrics> {
         self.base.compact(options).await
@@ -402,7 +404,7 @@ impl DatagenStore {
     /// Build a ZoneMap scalar index on `event_id`, the table's key column.
     /// Idempotent. Datagen previously had no scalar index, so every point
     /// lookup by event id scanned.
-    pub async fn create_event_id_index(&mut self) -> LanceResult<()> {
+    pub async fn create_event_id_index(&self) -> LanceResult<()> {
         self.base.create_key_zonemap_index().await
     }
 
@@ -433,7 +435,7 @@ impl DatagenStore {
                 let Some(store) = weak.upgrade() else {
                     return;
                 };
-                let mut guard = store.write().await;
+                let guard = store.write().await;
                 match tokio::time::timeout(pass_timeout, guard.cleanup_own_shard()).await {
                     Ok(Ok(0)) => {}
                     Ok(Ok(reclaimed)) => info!(
@@ -529,7 +531,7 @@ impl DatagenStore {
 
     fn non_blob_columns(&self) -> Vec<String> {
         self.base
-            .dataset
+            .current_dataset()
             .schema()
             .fields
             .iter()
@@ -1323,7 +1325,7 @@ mod tests {
         let uri = directory.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut writer_a = DatagenStore::open_with_options(
+            let writer_a = DatagenStore::open_with_options(
                 &uri,
                 DatagenStoreOptions {
                     storage_options: None,
@@ -1333,7 +1335,7 @@ mod tests {
             )
             .await
             .unwrap();
-            let mut writer_b = DatagenStore::open_with_options(
+            let writer_b = DatagenStore::open_with_options(
                 &uri,
                 DatagenStoreOptions {
                     storage_options: None,
@@ -1424,7 +1426,7 @@ mod tests {
         let uri = directory.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = DatagenStore::open(&uri).await.unwrap();
+            let store = DatagenStore::open(&uri).await.unwrap();
 
             // Root item "7" fans out into one sub-item "7/solve_twice:0".
             let mut root_created = event("7", 0, "created-root", 0, DatagenEventType::ItemCreated);
@@ -1478,7 +1480,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = DatagenStore::open(&uri).await.unwrap();
+            let store = DatagenStore::open(&uri).await.unwrap();
 
             // One cleanup pass appends one fragment, so merge after each append
             // to accumulate several -- this is exactly the growth pattern that
@@ -1519,7 +1521,7 @@ mod tests {
         let uri = directory.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = DatagenStore::open_with_options(
+            let store = DatagenStore::open_with_options(
                 &uri,
                 DatagenStoreOptions {
                     storage_options: None,
diff --git a/crates/lance-context-core/src/generic_store.rs b/crates/lance-context-core/src/generic_store.rs
index 4c55247..f3d6f5b 100644
--- a/crates/lance-context-core/src/generic_store.rs
+++ b/crates/lance-context-core/src/generic_store.rs
@@ -184,7 +184,7 @@ impl GenericStore {
         )
         .await?;
 
-        let schema: Arc<Schema> = Arc::new(base.dataset.schema().into());
+        let schema: Arc<Schema> = Arc::new(base.current_dataset().schema().into());
         let persisted = spec_from_schema(&schema)?;
 
         // Reopening with a different schema would reinterpret existing data.
@@ -221,7 +221,7 @@ impl GenericStore {
 
     /// URI of the underlying Lance dataset.
     #[must_use]
-    pub fn uri(&self) -> &str {
+    pub fn uri(&self) -> String {
         self.base.uri()
     }
 
@@ -238,7 +238,7 @@ impl GenericStore {
     }
 
     /// Refresh this handle to the latest base-table manifest.
-    pub async fn refresh_latest(&mut self) -> LanceResult<()> {
+    pub async fn refresh_latest(&self) -> LanceResult<()> {
         self.base.refresh_latest().await
     }
 
@@ -356,19 +356,19 @@ impl GenericStore {
     }
 
     /// Close the resident writer, draining its background tasks. Idempotent.
-    pub async fn close(&mut self) -> LanceResult<()> {
+    pub async fn close(&self) -> LanceResult<()> {
         self.base.close().await
     }
 
     /// Merge flushed generations into the base table once the count trigger is
     /// met. Returns how many were reclaimed.
-    pub async fn maybe_merge_wal(&mut self) -> LanceResult<usize> {
+    pub async fn maybe_merge_wal(&self) -> LanceResult<usize> {
         self.base.maybe_merge_own_shard().await
     }
 
     /// Seal, then merge **every** pending generation into the base table — the
     /// time half of the "time OR count" trigger.
-    pub async fn cleanup_wal(&mut self) -> LanceResult<usize> {
+    pub async fn cleanup_wal(&self) -> LanceResult<usize> {
         self.base.cleanup_own_shard().await
     }
 
@@ -380,7 +380,7 @@ impl GenericStore {
     /// Compact the base table's small fragments. Drive from a single external
     /// trigger, not per worker — see [`StorageBase::compact`].
     pub async fn compact(
-        &mut self,
+        &self,
         options: Option<CompactionConfig>,
     ) -> LanceResult<CompactionMetrics> {
         self.base.compact(options).await
@@ -399,14 +399,14 @@ impl GenericStore {
     }
 
     /// Build a ZoneMap scalar index on `id`. Idempotent.
-    pub async fn create_id_index(&mut self) -> LanceResult<()> {
+    pub async fn create_id_index(&self) -> LanceResult<()> {
         self.base.create_key_zonemap_index().await
     }
 
     /// Row count of the base table. Excludes rows still in unmerged
     /// generations or buffered in the writer.
     pub async fn count_base_rows(&self) -> LanceResult<usize> {
-        self.base.dataset.count_rows(None).await
+        self.base.current_dataset().count_rows(None).await
     }
 }
 
@@ -699,7 +699,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let rt = tokio::runtime::Runtime::new().unwrap();
         rt.block_on(async {
-            let mut store = GenericStore::open(&uri, spec(), sealing()).await.unwrap();
+            let store = GenericStore::open(&uri, spec(), sealing()).await.unwrap();
             let payload: Vec<u8> = (0..4 * 1024 * 1024).map(|i| (i % 251) as u8).collect();
             store
                 .add(&[row(json!({"id": "big", "payload": payload}))])
@@ -745,7 +745,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let rt = tokio::runtime::Runtime::new().unwrap();
         rt.block_on(async {
-            let mut store = GenericStore::open(&uri, spec(), sealing()).await.unwrap();
+            let store = GenericStore::open(&uri, spec(), sealing()).await.unwrap();
             for i in 0..3 {
                 store
                     .add(&[row(json!({"id": format!("r{i}")}))])
diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs
index 23c4149..979f60d 100644
--- a/crates/lance-context-core/src/rollout_store.rs
+++ b/crates/lance-context-core/src/rollout_store.rs
@@ -496,7 +496,7 @@ impl RolloutStore {
     }
     /// URI of the underlying Lance dataset.
     #[must_use]
-    pub fn uri(&self) -> &str {
+    pub fn uri(&self) -> String {
         self.base.uri()
     }
 
@@ -506,9 +506,11 @@ impl RolloutStore {
         self.base.version()
     }
 
-    /// Checkout a specific dataset version — recovers the exact rollout set that
-    /// trained a checkpoint (spec §3, reproducibility).
-    pub async fn checkout(&mut self, version_id: u64) -> LanceResult<()> {
+    /// Checkout a specific base-table version (time travel over the base table
+    /// only). Rollout training does not use this — reproduce a checkpoint by
+    /// filtering immutable rows (`policy_version`), not by pinning a dataset
+    /// version. See `docs/src/specs/rollout-deployment.md` §7.
+    pub async fn checkout(&self, version_id: u64) -> LanceResult<()> {
         self.base.checkout(version_id).await
     }
 
@@ -527,7 +529,7 @@ impl RolloutStore {
     /// Long-lived read handles call this before a new request so compaction or
     /// WAL merges committed by another process become visible without paying
     /// the cost of reopening the dataset and rebuilding all session caches.
-    pub async fn refresh_latest(&mut self) -> LanceResult<()> {
+    pub async fn refresh_latest(&self) -> LanceResult<()> {
         self.base.refresh_latest().await
     }
 
@@ -595,19 +597,20 @@ impl RolloutStore {
 
     /// Gracefully close the resident writer, draining its background tasks.
     /// Idempotent. See `StorageBase::close`.
-    pub async fn close(&mut self) -> LanceResult<()> {
+    pub async fn close(&self) -> LanceResult<()> {
         self.base.close().await
     }
 
     /// Merge this instance's flushed generations into the base table **if** the
     /// shard has accumulated at least `merge_after_generations` of them (the
     /// count trigger; `0` disables it). No-op otherwise.
-    pub async fn maybe_merge_own_shard(&mut self) -> LanceResult<usize> {
+    pub async fn maybe_merge_own_shard(&self) -> LanceResult<usize> {
         self.base.maybe_merge_own_shard().await
     }
 
-    /// The shared-lock half of a merge; see `StorageBase::prepare_merge_if_ready`
-    /// for the intended read-lock/write-lock split.
+    /// Prepare half of a merge; see `StorageBase::prepare_merge_if_ready`.
+    /// Merge exclusivity is the base's internal `merge_lock` (held inside the
+    /// returned [`PreparedMerge`]), not an outer store write lock.
     pub async fn prepare_merge_if_ready(
         &self,
         threshold: usize,
@@ -624,8 +627,10 @@ impl RolloutStore {
     }
 
     /// Commit a merge prepared by [`Self::prepare_merge_if_ready`].
+    ///
+    /// Consumes [`PreparedMerge`], releasing the base's merge lock when done.
     pub async fn commit_prepared_merge(
-        &mut self,
+        &self,
         manifest_store: &ShardManifestStore,
         manifest: &ShardManifest,
         prepared: PreparedMerge,
@@ -639,7 +644,7 @@ impl RolloutStore {
     /// then fold **every** pending flushed generation into the base table. This
     /// is the *time* half of the "time OR count" trigger and is deliberately not
     /// gated by the count threshold. See `StorageBase::cleanup_own_shard`.
-    pub async fn cleanup_own_shard(&mut self) -> LanceResult<usize> {
+    pub async fn cleanup_own_shard(&self) -> LanceResult<usize> {
         self.base.cleanup_own_shard().await
     }
 
@@ -649,7 +654,7 @@ impl RolloutStore {
     /// compaction rewrites the shared base table and two concurrent `Rewrite`
     /// commits conflict. See `StorageBase::compact`.
     pub async fn compact(
-        &mut self,
+        &self,
         options: Option<CompactionConfig>,
     ) -> LanceResult<CompactionMetrics> {
         self.base.compact(options).await
@@ -657,7 +662,7 @@ impl RolloutStore {
 
     /// Build a ZoneMap scalar index on the base table's `id` column. Idempotent.
     /// See `StorageBase::create_key_zonemap_index`.
-    pub async fn create_id_zonemap_index(&mut self) -> LanceResult<()> {
+    pub async fn create_id_zonemap_index(&self) -> LanceResult<()> {
         self.base.create_key_zonemap_index().await
     }
 
@@ -687,12 +692,17 @@ impl RolloutStore {
     /// writer owns and merges its own WAL shard.
     pub async fn observe(&self) -> LanceResult<RolloutObservation> {
         let shard_snapshots = self.wal_shard_snapshots().await?;
-        let base_rows = self.base.dataset.count_rows(None).await? as u64;
+        let base_rows = self.base.current_dataset().count_rows(None).await? as u64;
         let pending_rows = self.base.pending_wal_rows(&shard_snapshots).await?;
         let row_count = (base_rows + pending_rows) as i64;
-        let fragment_count = self.base.dataset.count_fragments() as i64;
-        let version = self.base.dataset.manifest.version;
-        let last_updated = self.base.dataset.manifest.timestamp().timestamp_millis();
+        let fragment_count = self.base.current_dataset().count_fragments() as i64;
+        let version = self.base.current_dataset().manifest.version;
+        let last_updated = self
+            .base
+            .current_dataset()
+            .manifest
+            .timestamp()
+            .timestamp_millis();
         let pending_wal_generations = shard_snapshots
             .iter()
             .map(|snapshot| snapshot.flushed_generations.len() as i64)
@@ -876,13 +886,13 @@ impl RolloutStore {
                         "pagination offset exceeds i64::MAX".to_string(),
                     ))
                 })?;
-                let mut scanner = self.base.dataset.scan();
+                let mut scanner = self.base.current_dataset().scan();
                 scanner.project(&refs)?;
                 // Lance 7's late take path can panic on nested list columns.
                 // Keep those early while deferring only potentially large text.
                 scanner.materialization_style(MaterializationStyle::all_early_except(
                     &PAGINATION_LATE_COLUMNS,
-                    self.base.dataset.schema(),
+                    self.base.current_dataset().schema(),
                 )?);
                 if let Some(filter) = &filter {
                     scanner.filter(filter)?;
@@ -978,14 +988,17 @@ impl RolloutStore {
         }
 
         let columns = Arc::new(self.non_blob_columns());
-        let target_schema = Arc::new(projected_arrow_schema(&self.base.dataset, &columns)?);
+        let target_schema = Arc::new(projected_arrow_schema(
+            self.base.current_dataset().as_ref(),
+            &columns,
+        )?);
         let id_filter = Arc::new(format!("id IN ({})", sql_quoted_list(&page_ids)));
         let wanted: HashSet<String> = page_ids.iter().cloned().collect();
 
         let mut records_by_id = HashMap::with_capacity(page_ids.len());
         if source == ListSource::All {
             for record in Self::take_page_rows_from_dataset(
-                self.base.dataset.clone(),
+                (*self.base.current_dataset()).clone(),
                 id_filter.clone(),
                 columns.clone(),
                 target_schema.clone(),
@@ -1125,7 +1138,7 @@ impl RolloutStore {
         let schema = match table_schema {
             Some(schema) => schema,
             None => {
-                let full: Schema = self.base.dataset.schema().into();
+                let full: Schema = self.base.current_dataset().schema().into();
                 let projected: Vec<FieldRef> = full
                     .fields()
                     .iter()
@@ -1234,7 +1247,7 @@ impl RolloutStore {
     ) -> LanceResult<Option<(RolloutRecord, Option<Vec<u8>>)>> {
         // Base table first — no manifest reads, no per-generation opens.
         if let Some(record) = self.scan_one_by_id(id, ListSource::Fragments).await? {
-            let payload = Self::get_blob_from_dataset(&self.base.dataset, id)
+            let payload = Self::get_blob_from_dataset(self.base.current_dataset().as_ref(), id)
                 .await?
                 .flatten();
             return Ok(Some((record, payload)));
@@ -1318,7 +1331,9 @@ impl RolloutStore {
     pub async fn get_blob(&self, id: &str) -> LanceResult<Option<Vec<u8>>> {
         // Base-table-first: an already-merged row is found here with no MemWAL
         // manifest reads and no per-generation opens.
-        if let Some(payload) = Self::get_blob_from_dataset(&self.base.dataset, id).await? {
+        if let Some(payload) =
+            Self::get_blob_from_dataset(self.base.current_dataset().as_ref(), id).await?
+        {
             return Ok(payload);
         }
 
@@ -1411,7 +1426,7 @@ impl RolloutStore {
     /// never materialize artifact bytes.
     fn non_blob_columns(&self) -> Vec<String> {
         self.base
-            .dataset
+            .current_dataset()
             .schema()
             .fields
             .iter()
@@ -1436,7 +1451,7 @@ impl RolloutStore {
     }
 
     fn records_to_batch(&self, records: &[RolloutRecord]) -> LanceResult<RecordBatch> {
-        let field_paths = self.base.dataset.schema().field_paths();
+        let field_paths = self.base.current_dataset().schema().field_paths();
         let has = |name: &str| field_paths.iter().any(|path| path == name);
         let include_relationships = has(RELATIONSHIPS_COLUMN);
         let include_metadata = has("metadata");
@@ -1691,7 +1706,7 @@ impl RolloutStore {
             arrays_by_name.insert("metadata".to_string(), Arc::new(metadata_builder.finish()));
         }
 
-        let schema: Arc<Schema> = Arc::new(self.base.dataset.schema().into());
+        let schema: Arc<Schema> = Arc::new(self.base.current_dataset().schema().into());
         let arrays = schema
             .fields()
             .iter()
@@ -2267,7 +2282,7 @@ mod tests {
         let legacy_schema = pre_claim_check_schema();
         create_empty_dataset(&uri, legacy_schema.clone()).await;
 
-        let mut store = RolloutStore::open_with_options(
+        let store = RolloutStore::open_with_options(
             &uri,
             RolloutStoreOptions {
                 shard_id: Some("pre-claim-check".to_string()),
@@ -2286,7 +2301,11 @@ mod tests {
             vec![Ok::<RecordBatch, ArrowError>(base_batch)].into_iter(),
             base_schema,
         );
-        store.base.dataset.append(base_reader, None).await.unwrap();
+        {
+            let mut dataset = (*store.base.current_dataset()).clone();
+            dataset.append(base_reader, None).await.unwrap();
+            store.base.set_dataset(dataset);
+        }
 
         store.add(&[assistant_record("legacy-wal")]).await.unwrap();
         store.flush().await.unwrap();
@@ -2300,16 +2319,18 @@ mod tests {
                 .filter(|field| legacy_schema.field_with_name(field.name()).is_err())
                 .cloned()
                 .collect::<Vec<_>>();
-            store
-                .base
-                .dataset
-                .add_columns(
-                    NewColumnTransform::AllNulls(Arc::new(Schema::new(claim_check_fields))),
-                    None,
-                    None,
-                )
-                .await
-                .unwrap();
+            {
+                let mut dataset = (*store.base.current_dataset()).clone();
+                dataset
+                    .add_columns(
+                        NewColumnTransform::AllNulls(Arc::new(Schema::new(claim_check_fields))),
+                        None,
+                        None,
+                    )
+                    .await
+                    .unwrap();
+                store.base.set_dataset(dataset);
+            }
         }
 
         (dir, store)
@@ -2574,7 +2595,7 @@ mod tests {
 
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = RolloutStore::open_with_options(
+            let store = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     storage_options: None,
@@ -2859,7 +2880,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = RolloutStore::open_with_options(
+            let store = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     shard_id: Some("fragment-pagination".to_string()),
@@ -2953,7 +2974,7 @@ mod tests {
             writer.add(&[assistant_record("row-0")]).await.unwrap();
             writer.flush().await.unwrap();
 
-            let mut cached_reader =
+            let cached_reader =
                 RolloutStore::open_existing_with_options(&uri, RolloutStoreOptions::default())
                     .await
                     .unwrap();
@@ -2983,7 +3004,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = RolloutStore::open(&uri).await.unwrap();
+            let store = RolloutStore::open(&uri).await.unwrap();
             assert!(!store.is_version_pinned());
 
             store.checkout(store.version()).await.unwrap();
@@ -3000,7 +3021,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = RolloutStore::open_with_options(
+            let store = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     shard_id: Some("trajectory-test".to_string()),
@@ -3050,8 +3071,13 @@ mod tests {
     /// Read the number of un-merged flushed generations recorded for a store's
     /// own write shard. Used by merge tests to assert the manifest drains.
     async fn flushed_generation_count(store: &RolloutStore) -> usize {
-        let object_store = store.base.dataset.object_store(None).await.unwrap();
-        let branch_location = store.base.dataset.branch_location();
+        let object_store = store
+            .base
+            .current_dataset()
+            .object_store(None)
+            .await
+            .unwrap();
+        let branch_location = store.base.current_dataset().branch_location();
         let manifest_store = ShardManifestStore::new(
             object_store,
             &branch_location.path,
@@ -3070,8 +3096,13 @@ mod tests {
     /// Used to assert the resident writer claims the epoch once instead of
     /// bumping it on every append.
     async fn shard_writer_epoch(store: &RolloutStore) -> u64 {
-        let object_store = store.base.dataset.object_store(None).await.unwrap();
-        let branch_location = store.base.dataset.branch_location();
+        let object_store = store
+            .base
+            .current_dataset()
+            .object_store(None)
+            .await
+            .unwrap();
+        let branch_location = store.base.current_dataset().branch_location();
         let manifest_store = ShardManifestStore::new(
             object_store,
             &branch_location.path,
@@ -3152,7 +3183,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = RolloutStore::open_with_options(
+            let store = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     storage_options: None,
@@ -3192,7 +3223,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = RolloutStore::open_with_options(
+            let store = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     storage_options: None,
@@ -3326,7 +3357,7 @@ mod tests {
         let artifact_bytes = b"\x00\x01\x02compacted";
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = RolloutStore::open_with_options(
+            let store = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     storage_options: None,
@@ -3355,7 +3386,7 @@ mod tests {
             store.flush().await.unwrap();
             store.maybe_merge_own_shard().await.unwrap();
 
-            let before = store.base.dataset.count_fragments();
+            let before = store.base.current_dataset().count_fragments();
             assert!(before > 1, "expected several fragments, got {before}");
             assert!(store.should_compact(&CompactionConfig {
                 min_fragments: 2,
@@ -3380,7 +3411,7 @@ mod tests {
                 "one incremental pass must honor max_source_fragments"
             );
 
-            let after = store.base.dataset.count_fragments();
+            let after = store.base.current_dataset().count_fragments();
             assert!(
                 after < before,
                 "compaction should reduce fragments: {before} -> {after}"
@@ -3426,7 +3457,7 @@ mod tests {
             .unwrap();
             store.add(&[assistant_record("a-0")]).await.unwrap();
 
-            let frags = store.base.dataset.count_fragments();
+            let frags = store.base.current_dataset().count_fragments();
             // A threshold above the current fragment count says "don't compact".
             assert!(!store.should_compact(&CompactionConfig {
                 min_fragments: frags + 1,
@@ -3458,7 +3489,7 @@ mod tests {
             .unwrap();
         runtime.block_on(async {
             // Seed the base table via A with several fragments to compact.
-            let mut a = RolloutStore::open_with_options(
+            let a = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     storage_options: None,
@@ -3499,11 +3530,11 @@ mod tests {
             // A compacts the base table; B merges its shard into it — concurrently.
             let (ca, mb) = tokio::join!(
                 async {
-                    let mut g = a.write().await;
+                    let g = a.write().await;
                     g.compact(None).await
                 },
                 async {
-                    let mut g = b.write().await;
+                    let g = b.write().await;
                     g.cleanup_own_shard().await
                 },
             );
@@ -3562,7 +3593,7 @@ mod tests {
         let artifact_bytes = b"\x00\x01\x02merged-trace";
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = RolloutStore::open_with_options(
+            let store = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     storage_options: None,
@@ -3619,7 +3650,7 @@ mod tests {
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
             {
-                let mut store = RolloutStore::open_with_options(
+                let store = RolloutStore::open_with_options(
                     &uri,
                     RolloutStoreOptions {
                         storage_options: None,
@@ -3659,7 +3690,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = RolloutStore::open_with_options(
+            let store = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     storage_options: None,
@@ -3698,11 +3729,11 @@ mod tests {
     fn cleanup_merges_pre_claim_check_generations_after_schema_evolution() {
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let (_dir, mut store) = store_with_legacy_base_and_wal(false).await;
+            let (_dir, store) = store_with_legacy_base_and_wal(false).await;
 
             assert_eq!(store.cleanup_own_shard().await.unwrap(), 1);
             assert_eq!(flushed_generation_count(&store).await, 0);
-            let field_paths = store.base.dataset.schema().field_paths();
+            let field_paths = store.base.current_dataset().schema().field_paths();
             for column in CLAIM_CHECK_COLUMNS {
                 assert!(field_paths.iter().any(|path| path == column));
             }
@@ -3729,7 +3760,7 @@ mod tests {
             let legacy_dir = TempDir::new().unwrap();
             let legacy_uri = legacy_dir.path().to_string_lossy().to_string();
             create_empty_dataset(&legacy_uri, pre_claim_check_schema()).await;
-            let mut legacy_store = RolloutStore::open(&legacy_uri).await.unwrap();
+            let legacy_store = RolloutStore::open(&legacy_uri).await.unwrap();
 
             let current_dir = TempDir::new().unwrap();
             let current_uri = current_dir.path().to_string_lossy().to_string();
@@ -3743,18 +3774,18 @@ mod tests {
             let generation_batch = current_store.records_to_batch(&[record]).unwrap();
 
             legacy_store.base.ensure_latest_schema().await.unwrap();
-            let merge_schema: Arc<Schema> = Arc::new(legacy_store.base.dataset.schema().into());
+            let merge_schema: Arc<Schema> =
+                Arc::new(legacy_store.base.current_dataset().schema().into());
             let aligned = align_batch_to_schema(generation_batch, merge_schema.clone()).unwrap();
             let reader = RecordBatchIterator::new(
                 vec![Ok::<RecordBatch, ArrowError>(aligned)].into_iter(),
                 merge_schema,
             );
-            legacy_store
-                .base
-                .dataset
-                .append(reader, None)
-                .await
-                .unwrap();
+            {
+                let mut dataset = (*legacy_store.base.current_dataset()).clone();
+                dataset.append(reader, None).await.unwrap();
+                legacy_store.base.set_dataset(dataset);
+            }
 
             let merged = legacy_store
                 .get_by_id_source("current-generation", ListSource::Fragments)
@@ -3815,7 +3846,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = RolloutStore::open(&uri).await.unwrap();
+            let store = RolloutStore::open(&uri).await.unwrap();
             // Add rows and fold them into the base table so there is data (and a
             // MemWAL index) present when we build the scalar index.
             store.add(&[assistant_record("a-0")]).await.unwrap();
@@ -3825,7 +3856,7 @@ mod tests {
 
             store.create_id_zonemap_index().await.unwrap();
             let has_id_index = |s: &RolloutStore| {
-                let dataset = s.base.dataset.clone();
+                let dataset = s.base.current_dataset();
                 async move {
                     dataset
                         .load_indices()
@@ -3900,11 +3931,11 @@ mod tests {
             // Both merge into the shared base table concurrently.
             let (ra, rb) = tokio::join!(
                 async {
-                    let mut g = a.write().await;
+                    let g = a.write().await;
                     g.cleanup_own_shard().await
                 },
                 async {
-                    let mut g = b.write().await;
+                    let g = b.write().await;
                     g.cleanup_own_shard().await
                 },
             );
@@ -3941,7 +3972,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = RolloutStore::open_with_options(
+            let store = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     storage_options: None,
@@ -3991,7 +4022,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = RolloutStore::open_with_options(
+            let store = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     storage_options: None,
@@ -4058,7 +4089,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = RolloutStore::open_with_options(
+            let store = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     shard_id: Some("rollout-pagination".to_string()),
@@ -4159,7 +4190,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut writer = RolloutStore::open_with_options(
+            let writer = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     shard_id: Some("pagination-benchmark".to_string()),
@@ -4338,7 +4369,7 @@ mod tests {
         runtime.block_on(async {
             // merge_after_generations = None: appended rows stay in the WAL,
             // un-merged, so this exercises the base-miss -> WAL-fallback path.
-            let mut store = RolloutStore::open_with_options(
+            let store = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     storage_options: None,
@@ -4421,7 +4452,7 @@ mod tests {
         let bytes = b"base-version-bytes";
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = RolloutStore::open_with_options(
+            let store = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     storage_options: None,
@@ -4460,7 +4491,7 @@ mod tests {
         let bytes = b"\x00\x01record-with-blob";
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = RolloutStore::open_with_options(
+            let store = RolloutStore::open_with_options(
                 &uri,
                 RolloutStoreOptions {
                     storage_options: None,
diff --git a/crates/lance-context-core/src/store.rs b/crates/lance-context-core/src/store.rs
index 2dc504b..748ab11 100644
--- a/crates/lance-context-core/src/store.rs
+++ b/crates/lance-context-core/src/store.rs
@@ -658,8 +658,8 @@ impl ContextStore {
 
     /// URI of the underlying Lance dataset.
     #[must_use]
-    pub fn uri(&self) -> &str {
-        self.base.dataset.uri()
+    pub fn uri(&self) -> String {
+        self.base.uri()
     }
 
     /// Distance metric this context ranks vector-search results with.
@@ -1374,7 +1374,7 @@ impl ContextStore {
 
     fn has_relationships_column(&self) -> bool {
         self.base
-            .dataset
+            .current_dataset()
             .schema()
             .field_paths()
             .iter()
@@ -1383,7 +1383,7 @@ impl ContextStore {
 
     fn has_external_id_column(&self) -> bool {
         self.base
-            .dataset
+            .current_dataset()
             .schema()
             .field_paths()
             .iter()
@@ -1399,22 +1399,26 @@ impl ContextStore {
     ///
     /// Existing rows are stored as null in the new column and read back as an
     /// empty relationship list.
-    pub async fn migrate_relationships_column(&mut self) -> LanceResult<bool> {
+    pub async fn migrate_relationships_column(&self) -> LanceResult<bool> {
         if self.has_relationships_column() {
             return Ok(false);
         }
 
         let schema = Arc::new(Schema::new(vec![relationship_field()]));
         self.base
-            .dataset
-            .add_columns(NewColumnTransform::AllNulls(schema), None, None)
+            .call_dataset_mut_fn(|mut dataset| async move {
+                dataset
+                    .add_columns(NewColumnTransform::AllNulls(schema), None, None)
+                    .await?;
+                Ok(dataset)
+            })
             .await?;
         self.base.clear_version_pin();
         Ok(true)
     }
 
     /// Checkout a specific dataset version.
-    pub async fn checkout(&mut self, version_id: u64) -> LanceResult<()> {
+    pub async fn checkout(&self, version_id: u64) -> LanceResult<()> {
         self.base.checkout(version_id).await
     }
 
@@ -1425,14 +1429,14 @@ impl ContextStore {
     }
 
     /// Refresh this handle to the latest base-table manifest.
-    pub async fn refresh_latest(&mut self) -> LanceResult<()> {
+    pub async fn refresh_latest(&self) -> LanceResult<()> {
         self.base.refresh_latest().await
     }
 
     /// Retrieve a single record by its unique ID.
     pub async fn get(&self, id: &str) -> LanceResult<Option<ContextRecord>> {
         let escaped_id = id.replace('\'', "''");
-        let mut scanner = self.base.dataset.scan();
+        let mut scanner = self.base.current_dataset().scan();
         scanner.filter(&format!("id = '{}'", escaped_id))?;
         scanner.limit(Some(1), None)?;
 
@@ -1835,7 +1839,7 @@ impl ContextStore {
     /// filtering and metadata stay correct).
     fn projected_columns(&self, projection: ReadProjection) -> Vec<String> {
         self.base
-            .dataset
+            .current_dataset()
             .schema()
             .fields
             .iter()
@@ -1887,14 +1891,14 @@ impl ContextStore {
 
     /// Manually trigger compaction to merge small fragments.
     pub async fn compact(
-        &mut self,
+        &self,
         options: Option<CompactionConfig>,
     ) -> LanceResult<CompactionMetrics> {
         let config = options.unwrap_or_else(|| self.compaction_config.clone());
 
         info!(
             "Starting compaction: {} fragments",
-            self.base.dataset.count_fragments()
+            self.base.current_dataset().count_fragments()
         );
         let start = std::time::Instant::now();
 
@@ -1970,7 +1974,7 @@ impl ContextStore {
 
     /// Gracefully close the resident MemWAL writer, draining its background
     /// tasks and sealing whatever it still buffers. Idempotent.
-    pub async fn close(&mut self) -> LanceResult<()> {
+    pub async fn close(&self) -> LanceResult<()> {
         self.base.close().await
     }
 
@@ -1984,14 +1988,14 @@ impl ContextStore {
     /// unioned all of them, so read cost grew without bound in the number of
     /// writes. Merging is what keeps that bounded — drive it from a sweeper, or
     /// use [`Self::cleanup_wal`] for the time-based trigger.
-    pub async fn maybe_merge_wal(&mut self) -> LanceResult<usize> {
+    pub async fn maybe_merge_wal(&self) -> LanceResult<usize> {
         self.base.maybe_merge_own_shard().await
     }
 
     /// Seal, then fold **every** pending flushed generation into the base table.
     /// The time half of the "time OR count" trigger, so deliberately not gated
     /// by the count threshold. Returns the number of generations reclaimed.
-    pub async fn cleanup_wal(&mut self) -> LanceResult<usize> {
+    pub async fn cleanup_wal(&self) -> LanceResult<usize> {
         self.base.cleanup_own_shard().await
     }
 
@@ -2003,7 +2007,7 @@ impl ContextStore {
 
     /// Check if compaction should run based on configuration thresholds.
     pub async fn should_compact(&self) -> LanceResult<bool> {
-        let fragment_count = self.base.dataset.count_fragments();
+        let fragment_count = self.base.current_dataset().count_fragments();
 
         if fragment_count < self.compaction_config.min_fragments {
             return Ok(false);
@@ -2030,7 +2034,7 @@ impl ContextStore {
         let state = self.compaction_state.lock().await;
 
         Ok(CompactionStats {
-            total_fragments: self.base.dataset.count_fragments(),
+            total_fragments: self.base.current_dataset().count_fragments(),
             is_compacting: state.is_compacting,
             last_compaction: state.last_compaction,
             last_error: state.last_error.clone(),
@@ -2039,12 +2043,12 @@ impl ContextStore {
     }
 
     /// Ensure the configured id index exists on the dataset.
-    async fn ensure_id_index(&mut self) -> LanceResult<()> {
+    async fn ensure_id_index(&self) -> LanceResult<()> {
         if self.id_index_type == IdIndexType::None {
             return Ok(());
         }
 
-        let indices = self.base.dataset.load_indices().await?;
+        let indices = self.base.current_dataset().load_indices().await?;
         if indices.iter().any(|i| i.name == ID_INDEX_NAME) {
             return Ok(());
         }
@@ -2053,7 +2057,7 @@ impl ContextStore {
     }
 
     /// Create (or replace) the scalar index on the `id` column.
-    pub async fn create_id_index(&mut self) -> LanceResult<()> {
+    pub async fn create_id_index(&self) -> LanceResult<()> {
         let index_type = match self.id_index_type {
             IdIndexType::ZoneMap => IndexType::ZoneMap,
             IdIndexType::BTree => IndexType::BTree,
@@ -2065,16 +2069,20 @@ impl ContextStore {
         let params = ScalarIndexParams::default();
 
         self.base
-            .dataset
-            .create_index_builder(&["id"], index_type, &params)
-            .name(ID_INDEX_NAME.to_string())
-            .replace(true)
-            .await?;
-
-        // Reload through the base so the new index is visible to subsequent
-        // reads, keeping the storage options and session (a bare
-        // `Dataset::open` here silently dropped them).
-        self.base.reload().await
+            .with_exclusive_writer(|| async {
+                self.base
+                    .call_dataset_mut_fn(|mut dataset| async move {
+                        dataset
+                            .create_index_builder(&["id"], index_type, &params)
+                            .name(ID_INDEX_NAME.to_string())
+                            .replace(true)
+                            .await?;
+                        Ok(dataset)
+                    })
+                    .await?;
+                self.base.reload().await
+            })
+            .await
     }
 
     /// Start background compaction task if enabled.
@@ -2100,7 +2108,7 @@ impl ContextStore {
         // have exactly one owner. A second handle is the right model anyway --
         // compaction only rewrites base-table fragments and takes `&mut`, so
         // sharing a handle with the write path would mean contending for it.
-        let uri = self.uri().to_string();
+        let uri = self.uri();
         let interval_secs = self.compaction_config.check_interval_secs;
         let options = ContextStoreOptions {
             storage_options: self.base.storage_options.clone(),
@@ -2131,7 +2139,7 @@ impl ContextStore {
             loop {
                 interval.tick().await;
 
-                let mut store = match open_for_compaction(&uri, compaction_options.clone()).await {
+                let store = match open_for_compaction(&uri, compaction_options.clone()).await {
                     Ok(store) => store,
                     Err(e) => {
                         error!("Background compaction could not open store: {}", e);
@@ -2338,42 +2346,42 @@ impl ContextStore {
     fn records_to_batch(&self, entries: &[ContextRecord]) -> LanceResult<RecordBatch> {
         let include_external_id = self
             .base
-            .dataset
+            .current_dataset()
             .schema()
             .field_paths()
             .iter()
             .any(|path| path == "external_id");
         let include_lifecycle = self
             .base
-            .dataset
+            .current_dataset()
             .schema()
             .field_paths()
             .iter()
             .any(|path| path == "expires_at");
         let include_metadata = self
             .base
-            .dataset
+            .current_dataset()
             .schema()
             .field_paths()
             .iter()
             .any(|path| path == "metadata");
         let include_tenant = self
             .base
-            .dataset
+            .current_dataset()
             .schema()
             .field_paths()
             .iter()
             .any(|path| path == "tenant");
         let include_source = self
             .base
-            .dataset
+            .current_dataset()
             .schema()
             .field_paths()
             .iter()
             .any(|path| path == "source");
         let include_external_reference = self
             .base
-            .dataset
+            .current_dataset()
             .schema()
             .field_paths()
             .iter()
@@ -2698,7 +2706,7 @@ impl ContextStore {
             ]);
         }
 
-        let schema: Arc<Schema> = Arc::new(self.base.dataset.schema().into());
+        let schema: Arc<Schema> = Arc::new(self.base.current_dataset().schema().into());
         let arrays = schema
             .fields()
             .iter()
@@ -4471,7 +4479,7 @@ mod tests {
             .await
             .unwrap();
 
-            let mut store = ContextStore::open(&uri).await.unwrap();
+            let store = ContextStore::open(&uri).await.unwrap();
             assert!(!store.has_relationships_column());
 
             let mut record = text_record("with-relationships", 0.0);
@@ -5046,7 +5054,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let runtime = tokio::runtime::Runtime::new().unwrap();
         runtime.block_on(async {
-            let mut store = ContextStore::open(&uri).await.unwrap();
+            let store = ContextStore::open(&uri).await.unwrap();
             for i in 0..3 {
                 store
                     .add(&[text_record(&format!("r{i}"), i as f32)])
@@ -5556,12 +5564,12 @@ mod tests {
                 id_index_type: IdIndexType::BTree,
                 ..Default::default()
             };
-            let mut store = ContextStore::open_with_options(&uri, options)
+            let store = ContextStore::open_with_options(&uri, options)
                 .await
                 .unwrap();
 
             // Index should be created eagerly on open
-            let indices = store.base.dataset.load_indices().await.unwrap();
+            let indices = store.base.current_dataset().load_indices().await.unwrap();
             assert!(
                 indices.iter().any(|i| i.name == ID_INDEX_NAME),
                 "btree index should be created on open"
@@ -5577,7 +5585,7 @@ mod tests {
             store.compact(None).await.unwrap();
 
             // Index should still exist after compaction
-            let indices = store.base.dataset.load_indices().await.unwrap();
+            let indices = store.base.current_dataset().load_indices().await.unwrap();
             assert!(
                 indices.iter().any(|i| i.name == ID_INDEX_NAME),
                 "btree index should persist after compaction"
@@ -5596,12 +5604,12 @@ mod tests {
                 id_index_type: IdIndexType::ZoneMap,
                 ..Default::default()
             };
-            let mut store = ContextStore::open_with_options(&uri, options)
+            let store = ContextStore::open_with_options(&uri, options)
                 .await
                 .unwrap();
 
             // Index should be created eagerly on open
-            let indices = store.base.dataset.load_indices().await.unwrap();
+            let indices = store.base.current_dataset().load_indices().await.unwrap();
             assert!(
                 indices.iter().any(|i| i.name == ID_INDEX_NAME),
                 "zonemap index should be created on open"
@@ -5615,7 +5623,7 @@ mod tests {
             }
             store.compact(None).await.unwrap();
 
-            let indices = store.base.dataset.load_indices().await.unwrap();
+            let indices = store.base.current_dataset().load_indices().await.unwrap();
             assert!(
                 indices.iter().any(|i| i.name == ID_INDEX_NAME),
                 "zonemap index should persist after compaction"
@@ -5630,12 +5638,12 @@ mod tests {
         let runtime = tokio::runtime::Runtime::new().unwrap();
 
         runtime.block_on(async {
-            let mut store = ContextStore::open(&uri).await.unwrap();
+            let store = ContextStore::open(&uri).await.unwrap();
 
             store.add(&[text_record("no-idx-1", 0.0)]).await.unwrap();
             store.compact(None).await.unwrap();
 
-            let indices = store.base.dataset.load_indices().await.unwrap();
+            let indices = store.base.current_dataset().load_indices().await.unwrap();
             assert!(
                 !indices.iter().any(|i| i.name == ID_INDEX_NAME),
                 "no id index should be created when IdIndexType::None"
@@ -5654,7 +5662,7 @@ mod tests {
                 id_index_type: IdIndexType::BTree,
                 ..Default::default()
             };
-            let mut store = ContextStore::open_with_options(&uri, options)
+            let store = ContextStore::open_with_options(&uri, options)
                 .await
                 .unwrap();
 
diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs
index b534a22..aa2cd4c 100644
--- a/crates/lance-context-core/src/store_base.rs
+++ b/crates/lance-context-core/src/store_base.rs
@@ -27,6 +27,20 @@
 //! - **Every dataset open goes through [`StorageBase::load_with_options`]**, so
 //!   storage options and the shared session are never silently dropped.
 //!
+//! # Dataset-handle publish
+//!
+//! Readers `ArcSwap::load` a snapshot and never take a writer lock. Every
+//! replacement of the handle takes [`StorageBase::write_writer`]:
+//!
+//! - **Single RMW** ([`StorageBase::call_dataset_mut_fn`],
+//!   [`StorageBase::call_dataset_with`]): lock → mutate → `set_dataset`.
+//! - **Multi-step RMW** ([`StorageBase::with_exclusive_writer`]): one lock
+//!   covers several publishes so a concurrent `checkout` cannot land between
+//!   them. Nesting is task-local; another task always waits on the mutex.
+//!
+//! Steady-state MemWAL `put`s clone the resident `ShardWriter` and do **not**
+//! hold this lock; only first-open / fence-reopen and handle publish do.
+//!
 //! # What stays in the concrete store
 //!
 //! Anything that needs to know the schema: the Arrow schema itself,
@@ -36,8 +50,24 @@
 //! latest schema to evolve a base table to — via [`StorageBaseOptions`].
 
 use std::collections::{HashMap, HashSet};
-use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
+
+tokio::task_local! {
+    /// Set while *this task* holds [`StorageBase::write_writer`] for exclusive
+    /// handle publish / nested RMW.
+    ///
+    /// Must be task-local: a shared `AtomicBool` would let a *different* task
+    /// observe "held" and skip the mutex, defeating the critical section.
+    static WRITE_WRITER_HELD: ();
+}
 
+#[inline]
+fn exclusive_writer_held_here() -> bool {
+    WRITE_WRITER_HELD.try_with(|_| ()).is_ok()
+}
+
+use arc_swap::ArcSwap;
 use arrow_array::{new_null_array, RecordBatch, RecordBatchIterator, UInt32Array};
 use arrow_schema::{ArrowError, Schema};
 use arrow_select::take::take;
@@ -180,11 +210,12 @@ pub enum ListSource {
 /// Rows read out of the flushed generations, ready to be appended to the base
 /// table and drained from the shard manifest.
 ///
-/// Produced by `StorageBase::prepare_merge_if_ready` under `&self` (so appends
-/// keep running while it reads object storage) and consumed by
-/// `StorageBase::commit_prepared_merge` under `&mut self`. `PreparedMerge` is
-/// public because it appears in [`RolloutStore`]'s prepare/commit split, but
-/// its fields are opaque.
+/// Produced by `StorageBase::prepare_merge_if_ready` / `prepare_cleanup_merge`
+/// and consumed by `StorageBase::commit_prepared_merge`. Holding a value of this
+/// type retains the store's internal merge lock until it is committed or
+/// dropped, so callers of the prepare/commit split do not need to lock
+/// externally. `PreparedMerge` is public because it appears in
+/// [`RolloutStore`]'s prepare/commit split, but its fields are opaque.
 ///
 /// [`RolloutStore`]: crate::RolloutStore
 pub struct PreparedMerge {
@@ -192,6 +223,9 @@ pub struct PreparedMerge {
     merged_paths: Vec<String>,
     batches: Vec<RecordBatch>,
     merge_schema: Arc<Schema>,
+    /// Serializes this prepare+commit against other merges on the same store.
+    /// Released when `PreparedMerge` is dropped (after commit or on abandon).
+    _merge_guard: tokio::sync::OwnedMutexGuard<()>,
 }
 
 impl PreparedMerge {
@@ -259,12 +293,23 @@ pub(crate) struct StorageBaseOptions {
     pub seal_on_put: bool,
 }
 
+/// Per-handle compaction counters/timestamps.
+#[derive(Debug, Default)]
+struct CompactionState {
+    last_compaction: Option<DateTime<Utc>>,
+    total_compactions: u64,
+    last_error: Option<String>,
+}
+
 /// Schema-agnostic Lance storage: dataset handle, MemWAL write path, WAL merge,
 /// compaction, indexing, and LSM reads. See the module docs.
 pub(crate) struct StorageBase {
     /// The base table. `pub(crate)` because concrete stores build their own
     /// schema-specific scans and projections directly against it.
-    pub dataset: Dataset,
+    ///
+    /// Wrapped in [`ArcSwap`] so a merge/compact/reload can publish a new
+    /// handle without requiring exclusive `&mut` for every reader.
+    pub dataset: ArcSwap<Dataset>,
     /// MemWAL shard this instance writes to (derived from `shard_id`).
     pub write_shard: Uuid,
     /// Object-store options, retained so a self-merge can re-append flushed
@@ -283,23 +328,35 @@ pub(crate) struct StorageBase {
     /// Self-merge threshold; `0` disables it.
     merge_after_generations: usize,
     merge_max_generations: usize,
-    /// Timestamp of the last successful [`Self::compact`] on this handle.
-    last_compaction: Option<DateTime<Utc>>,
-    /// Number of successful compactions performed by this handle.
-    total_compactions: u64,
-    /// Error message from the most recent failed compaction on this handle.
-    last_compaction_error: Option<String>,
+    /// Compaction bookkeeping for this handle (interior-mutable so [`Self::compact`] can be `&self`).
+    compaction: Mutex<CompactionState>,
+    /// Serializes WAL→base merge (prepare through commit). Taken with
+    /// `try_lock_owned`: a loser no-ops (`Ok(0)` / `Ok(None)`). Not held by
+    /// `add`/`flush`, so appends keep running while a merge is in flight.
+    ///
+    /// Does **not** alone protect [`Self::set_dataset`]. Handle publish is
+    /// exclusive under [`Self::write_writer`]. Lock order: this lock, then
+    /// `write_writer`. Never acquire this while holding `write_writer`.
+    merge_lock: Arc<tokio::sync::Mutex<()>>,
     /// Explicit time-travel version selected by [`Self::checkout`].
     ///
     /// A point-read miss may refresh an ordinary long-lived handle to avoid a
     /// false negative from a stale manifest, but must never advance a handle
     /// whose caller deliberately selected a historical version.
-    pinned_version: Option<u64>,
+    ///
+    /// `0` means unpinned; any other value is the pinned manifest version.
+    /// (Lance dataset versions are 1-based, so `0` is never a real pin.)
+    pinned_version: AtomicU64,
     /// Resident MemWAL writer for this instance's shard, wrapped for `&self`
-    /// concurrent access. The [`tokio::sync::Mutex`] is held only to
-    /// fetch-or-open and clone the `Arc` (see [`Self::resident_writer`]) and to
-    /// invalidate a fenced writer (see [`Self::invalidate_writer`]); it is
-    /// **never** held across `put`, so steady-state appends run concurrently.
+    /// concurrent access.
+    ///
+    /// Also the exclusive lock for every dataset-handle RMW (`checkout`,
+    /// `refresh_latest`, merge append, compact, reload, schema/index). Held
+    /// only to fetch-or-open / invalidate the `ShardWriter` (see
+    /// [`Self::resident_writer`]) or across a handle publish — **never** across
+    /// `put`, so steady-state appends run concurrently. Multi-step publishers
+    /// use [`Self::with_exclusive_writer`]; nesting is task-local so only the
+    /// holding task skips re-acquire.
     write_writer: tokio::sync::Mutex<Option<Arc<ShardWriter>>>,
 }
 
@@ -402,8 +459,8 @@ impl StorageBase {
             .into());
         }
 
-        let mut base = Self {
-            dataset,
+        let base = Self {
+            dataset: ArcSwap::from_pointee(dataset),
             write_shard: derive_shard_id(shard_id.as_deref()),
             storage_options,
             session,
@@ -412,41 +469,57 @@ impl StorageBase {
             seal_on_put,
             merge_after_generations: merge_after_generations.unwrap_or(0),
             merge_max_generations: merge_max_generations.unwrap_or(DEFAULT_MERGE_MAX_GENERATIONS),
-            last_compaction: None,
-            total_compactions: 0,
-            last_compaction_error: None,
-            pinned_version: None,
+            compaction: Mutex::new(CompactionState::default()),
+            merge_lock: Arc::new(tokio::sync::Mutex::new(())),
+            pinned_version: AtomicU64::new(0),
             write_writer: tokio::sync::Mutex::new(None),
         };
         // `ensure_mem_wal` may reload the dataset on a concurrent first-writer
-        // race, which is why it must run here where we hold `&mut`.
+        // race; it publishes the new handle via ArcSwap.
         base.ensure_mem_wal().await?;
         Ok(base)
     }
 
     /// URI of the underlying Lance dataset.
     #[must_use]
-    pub fn uri(&self) -> &str {
-        self.dataset.uri()
+    pub fn uri(&self) -> String {
+        self.current_dataset().uri().to_string()
     }
 
     /// Current base dataset manifest version.
     #[must_use]
     pub fn version(&self) -> u64 {
-        self.dataset.manifest.version
+        self.current_dataset().manifest.version
     }
 
     /// Check out a specific base dataset version (time travel).
-    pub async fn checkout(&mut self, version_id: u64) -> LanceResult<()> {
-        self.dataset = self.dataset.checkout_version(version_id).await?;
-        self.pinned_version = Some(version_id);
+    ///
+    /// `version_id` must be non-zero (`0` is reserved to mean "unpinned").
+    ///
+    /// Lance's `checkout_version` takes `&self` and returns a **new** `Dataset`
+    /// handle aimed at that manifest (same URI/session, different view) — it
+    /// does not mutate the caller's value in place. Publishing is exclusive
+    /// under [`Self::write_writer`] so an older version can actually be pinned.
+    pub async fn checkout(&self, version_id: u64) -> LanceResult<()> {
+        if version_id == 0 {
+            return Err(ArrowError::InvalidArgumentError(
+                "dataset version 0 is reserved; pin a real manifest version (>= 1)".to_string(),
+            )
+            .into());
+        }
+        self.call_dataset_with(|current| async move {
+            let dataset = current.checkout_version(version_id).await?;
+            Ok((dataset, ()))
+        })
+        .await?;
+        self.pinned_version.store(version_id, Ordering::Release);
         Ok(())
     }
 
     /// Whether this handle was explicitly checked out to a historical version.
     #[must_use]
     pub fn is_version_pinned(&self) -> bool {
-        self.pinned_version.is_some()
+        self.pinned_version.load(Ordering::Acquire) != 0
     }
 
     /// Refresh this handle to the latest base-table manifest while retaining its
@@ -455,16 +528,111 @@ impl StorageBase {
     /// Long-lived read handles call this before a new request so compaction or
     /// WAL merges committed by another process become visible without paying the
     /// cost of reopening the dataset and rebuilding all session caches.
-    pub async fn refresh_latest(&mut self) -> LanceResult<()> {
-        self.dataset.checkout_latest().await?;
-        self.pinned_version = None;
+    pub async fn refresh_latest(&self) -> LanceResult<()> {
+        self.call_dataset_mut_fn(|mut dataset| async move {
+            dataset.checkout_latest().await?;
+            Ok(dataset)
+        })
+        .await?;
+        self.clear_version_pin();
         Ok(())
     }
 
     /// Mark this handle as no longer pinned after a concrete store mutates the
     /// dataset directly.
-    pub(crate) fn clear_version_pin(&mut self) {
-        self.pinned_version = None;
+    pub(crate) fn clear_version_pin(&self) {
+        self.pinned_version.store(0, Ordering::Release);
+    }
+
+    /// Current dataset snapshot (`Arc` clone; cheap).
+    #[inline]
+    pub(crate) fn current_dataset(&self) -> Arc<Dataset> {
+        self.dataset.load_full()
+    }
+
+    /// Publish a replacement dataset handle after a mutating Lance op.
+    ///
+    /// Unconditional `store`. Prefer [`Self::call_dataset_with`] /
+    /// [`Self::call_dataset_mut_fn`].
+    #[inline]
+    pub(crate) fn set_dataset(&self, dataset: Dataset) {
+        self.dataset.store(Arc::new(dataset));
+    }
+
+    /// Build a new handle from the current one (or ignore it — e.g.
+    /// [`Self::reload`]) and publish it under [`Self::write_writer`].
+    ///
+    /// Nested publishes from [`Self::with_exclusive_writer`] reuse that section
+    /// instead of acquiring the lock again.
+    ///
+    /// `f` receives the current `Arc<Dataset>` and returns `(new_handle, out)`.
+    /// The handle is published only if `f` succeeds. Do not call
+    /// [`Self::reload`] (or anything else that takes `write_writer`) from
+    /// inside `f` — the mutex is not reentrant.
+    pub(crate) async fn call_dataset_with<F, Fut, T>(&self, f: F) -> LanceResult<T>
+    where
+        F: FnOnce(Arc<Dataset>) -> Fut,
+        Fut: std::future::Future<Output = LanceResult<(Dataset, T)>>,
+    {
+        if exclusive_writer_held_here() {
+            return self.call_dataset_with_locked(f).await;
+        }
+        let _guard = self.write_writer.lock().await;
+        WRITE_WRITER_HELD
+            .scope((), self.call_dataset_with_locked(f))
+            .await
+    }
+
+    async fn call_dataset_with_locked<F, Fut, T>(&self, f: F) -> LanceResult<T>
+    where
+        F: FnOnce(Arc<Dataset>) -> Fut,
+        Fut: std::future::Future<Output = LanceResult<(Dataset, T)>>,
+    {
+        let current = self.current_dataset();
+        let (dataset, out) = f(current).await?;
+        self.before_publish_dataset().await;
+        self.set_dataset(dataset);
+        Ok(out)
+    }
+
+    /// Clone the current handle, run `f`, and publish the result.
+    ///
+    /// Single-step RMW (`append`, `checkout_latest`, `add_columns`, …). For
+    /// several publishes in one critical section, wrap them in
+    /// [`Self::with_exclusive_writer`].
+    pub(crate) async fn call_dataset_mut_fn<F, Fut>(&self, f: F) -> LanceResult<()>
+    where
+        F: FnOnce(Dataset) -> Fut,
+        Fut: std::future::Future<Output = LanceResult<Dataset>>,
+    {
+        self.call_dataset_with(|current| async move {
+            let dataset = f((*current).clone()).await?;
+            Ok((dataset, ()))
+        })
+        .await
+    }
+
+    /// Run `f` under one exclusive `write_writer` section so multiple handle
+    /// publishes compose.
+    ///
+    /// Nesting is allowed only for the **same task** (via
+    /// [`WRITE_WRITER_HELD`]); other tasks block on the mutex.
+    pub(crate) async fn with_exclusive_writer<F, Fut, T>(&self, f: F) -> LanceResult<T>
+    where
+        F: FnOnce() -> Fut,
+        Fut: std::future::Future<Output = LanceResult<T>>,
+    {
+        if exclusive_writer_held_here() {
+            return f().await;
+        }
+        let _guard = self.write_writer.lock().await;
+        WRITE_WRITER_HELD.scope((), f()).await
+    }
+
+    /// Test-only pause point after modify, before the handle is published.
+    async fn before_publish_dataset(&self) {
+        #[cfg(test)]
+        dataset_rmw_test_hooks::await_before_publish().await;
     }
 
     // ---------------------------------------------------------------- writes
@@ -560,7 +728,7 @@ impl StorageBase {
             ..Default::default()
         };
         let writer = Arc::new(
-            self.dataset
+            self.current_dataset()
                 .mem_wal_writer(self.write_shard, config)
                 .await?,
         );
@@ -641,9 +809,9 @@ impl StorageBase {
     /// by an explicit `close().await`. Call this before dropping a store on a
     /// path that can `await` (e.g. an LRU eviction that owns the last handle).
     /// Idempotent: a no-op when no writer is resident.
-    pub async fn close(&mut self) -> LanceResult<()> {
-        // `&mut self` gives exclusive access, so `get_mut` avoids an async lock.
-        if let Some(writer) = self.write_writer.get_mut().take() {
+    pub async fn close(&self) -> LanceResult<()> {
+        let writer = self.write_writer.lock().await.take();
+        if let Some(writer) = writer {
             match Arc::try_unwrap(writer) {
                 // Sole owner: drain the writer's background tasks gracefully.
                 Ok(writer) => writer.close().await?,
@@ -681,7 +849,7 @@ impl StorageBase {
     /// Merge this instance's flushed generations into the base table **if** the
     /// shard has accumulated at least `merge_after_generations` of them (the
     /// count trigger; `0` disables it). No-op otherwise.
-    pub async fn maybe_merge_own_shard(&mut self) -> LanceResult<usize> {
+    pub async fn maybe_merge_own_shard(&self) -> LanceResult<usize> {
         if self.merge_after_generations == 0 {
             return Ok(0);
         }
@@ -704,7 +872,7 @@ impl StorageBase {
     /// would stay empty, so the threshold check would return `0` and never reach
     /// the merge — leaving rows durable but permanently invisible until a
     /// process restart replayed the WAL.
-    pub async fn cleanup_own_shard(&mut self) -> LanceResult<usize> {
+    pub async fn cleanup_own_shard(&self) -> LanceResult<usize> {
         self.flush().await?;
         // Threshold `1`: merge whenever at least one generation is pending. The
         // time trigger must not depend on the count threshold — that is what
@@ -712,7 +880,7 @@ impl StorageBase {
         self.merge_own_shard_if_ready(1).await
     }
 
-    async fn merge_own_shard_if_ready(&mut self, threshold: usize) -> LanceResult<usize> {
+    async fn merge_own_shard_if_ready(&self, threshold: usize) -> LanceResult<usize> {
         let Some((manifest_store, manifest, prepared)) =
             self.prepare_merge_if_ready(threshold).await?
         else {
@@ -725,18 +893,18 @@ impl StorageBase {
         Ok(if committed { pending } else { 0 })
     }
 
-    /// The shared-lock half of a merge: decide whether one is due and read the
-    /// flushed generations into memory.
+    /// Prepare half of a merge: decide whether one is due and read the flushed
+    /// generations into memory.
     ///
-    /// Takes `&self`, so a caller holding a *read* lock can run the expensive
-    /// part while appends continue, then take the write lock only to hand the
-    /// result to [`Self::commit_prepared_merge`]. Returns `None` when nothing is
-    /// due.
+    /// Acquires the internal merge lock with `try_lock` (returned inside
+    /// [`PreparedMerge`]) so prepare+commit stay exclusive without blocking
+    /// `add`. Returns `None` when nothing is due **or** another merge already
+    /// holds the lock.
     ///
     /// ```ignore
-    /// let prepared = { store.read().await.prepare_merge_if_ready(1).await? };
+    /// let prepared = store.prepare_merge_if_ready(1).await?;
     /// if let Some((manifest_store, manifest, prepared)) = prepared {
-    ///     store.write().await.commit_prepared_merge(&manifest_store, &manifest, prepared).await?;
+    ///     store.commit_prepared_merge(&manifest_store, &manifest, prepared).await?;
     /// }
     /// ```
     pub async fn prepare_merge_if_ready(
@@ -761,12 +929,19 @@ impl StorageBase {
         threshold: usize,
         seal_first: bool,
     ) -> LanceResult<Option<(ShardManifestStore, ShardManifest, PreparedMerge)>> {
+        // Exclusive for the whole prepare→commit lifetime (guard lives in
+        // PreparedMerge). Losers no-op: the holder will drain current gens.
+        let Ok(merge_guard) = Arc::clone(&self.merge_lock).try_lock_owned() else {
+            return Ok(None);
+        };
+
         if seal_first {
             // Materialize anything buffered so it is eligible for this pass.
             self.flush().await?;
         }
-        let object_store = self.dataset.object_store(None).await?;
-        let branch_location = self.dataset.branch_location();
+        let dataset = self.current_dataset();
+        let object_store = dataset.object_store(None).await?;
+        let branch_location = dataset.branch_location();
         let manifest_store = ShardManifestStore::new(
             object_store,
             &branch_location.path,
@@ -780,15 +955,17 @@ impl StorageBase {
         if pending == 0 || pending < threshold.max(1) {
             return Ok(None);
         }
-        let Some(prepared) = self.prepare_merge(&manifest).await? else {
+        let Some(prepared) = self.prepare_merge(&manifest, merge_guard).await? else {
             return Ok(None);
         };
         Ok(Some((manifest_store, manifest, prepared)))
     }
 
     /// Commit a merge prepared by [`Self::prepare_merge_if_ready`].
+    ///
+    /// Consumes [`PreparedMerge`], releasing the merge lock when it returns.
     pub async fn commit_prepared_merge(
-        &mut self,
+        &self,
         manifest_store: &ShardManifestStore,
         manifest: &ShardManifest,
         prepared: PreparedMerge,
@@ -800,21 +977,21 @@ impl StorageBase {
         Ok(if committed { pending } else { 0 })
     }
 
-    /// The `&self` half of a merge: everything that can run while appends
-    /// continue — sealing the memtable and reading every flushed generation
-    /// into memory.
+    /// Seal + read every flushed generation into a [`PreparedMerge`].
     ///
-    /// # Concurrency: the expensive phase does not need exclusive access
+    /// # Concurrency
     ///
-    /// A merge only ever touches *sealed* generations — history — while a `put`
-    /// writes the active memtable at the WAL tail. They operate on disjoint
-    /// data, which is the whole point of an LSM, so a merge must not stop the
-    /// write path. Notably the merge does **not** `claim_epoch`: the epoch is an
-    /// *ownership* token, not a per-commit token, and
+    /// Caller already holds [`Self::merge_lock`] via `merge_guard`. A merge only
+    /// ever touches *sealed* generations — history — while a `put` writes the
+    /// active memtable at the WAL tail, so appends keep running. The merge does
+    /// **not** `claim_epoch`: the epoch is an *ownership* token, and
     /// [`ShardManifestStore::commit_update`] only rejects a writer whose epoch is
-    /// **older** than the stored one. Reusing the shard's current epoch commits
-    /// the drain and leaves the live writer untouched.
-    async fn prepare_merge(&self, manifest: &ShardManifest) -> LanceResult<Option<PreparedMerge>> {
+    /// **older** than the stored one.
+    async fn prepare_merge(
+        &self,
+        manifest: &ShardManifest,
+        merge_guard: tokio::sync::OwnedMutexGuard<()>,
+    ) -> LanceResult<Option<PreparedMerge>> {
         if manifest.flushed_generations.is_empty() {
             return Ok(None);
         }
@@ -825,8 +1002,8 @@ impl StorageBase {
         observe_phase!("seal", self.flush().await)?;
 
         // The expensive phase: pull every flushed generation out of object
-        // storage. Buffered in memory, so this is the part that must not hold an
-        // exclusive lock.
+        // storage. Runs under the merge lock so a second merge cannot prepare
+        // the same generations concurrently.
         let (merged_generations, merged_paths, batches, merge_schema) =
             observe_phase!("read", self.read_flushed_generations(manifest).await)?;
 
@@ -835,12 +1012,12 @@ impl StorageBase {
             merged_paths,
             batches,
             merge_schema,
+            _merge_guard: merge_guard,
         }))
     }
 
-    /// The `&mut self` half of a merge: append the prepared rows to the base
-    /// table, drain the merged generations from the manifest, and delete their
-    /// directories.
+    /// Append the prepared rows to the base table, drain the merged generations
+    /// from the manifest, and delete their directories.
     ///
     /// # Surgical drain, not blanket clear
     ///
@@ -861,7 +1038,7 @@ impl StorageBase {
     /// of appending physical duplicates. The next attempt can then drain the
     /// manifest without relying on a particular Lance read-plan shape.
     async fn commit_merge(
-        &mut self,
+        &self,
         manifest_store: &ShardManifestStore,
         manifest: &ShardManifest,
         prepared: PreparedMerge,
@@ -871,6 +1048,7 @@ impl StorageBase {
             merged_paths,
             batches,
             merge_schema,
+            _merge_guard,
         } = prepared;
 
         // Several sweepers can prepare the same immutable generations under a
@@ -887,15 +1065,19 @@ impl StorageBase {
             return Ok(false);
         }
 
-        self.ensure_latest_schema().await?;
+        self.with_exclusive_writer(|| async {
+            self.ensure_latest_schema().await?;
 
-        if !batches.is_empty() {
-            observe_phase!(
-                "append",
-                self.merge_prepared_batches(batches, merge_schema).await
-            )?;
-            self.pinned_version = None;
-        }
+            if !batches.is_empty() {
+                observe_phase!(
+                    "append",
+                    self.merge_prepared_batches(batches, merge_schema).await
+                )?;
+                self.clear_version_pin();
+            }
+            Ok(())
+        })
+        .await?;
 
         // Reuse the shard's *current* epoch rather than claiming a new one:
         // claiming would fence our own live writer. `commit_update` still fails
@@ -923,7 +1105,7 @@ impl StorageBase {
         )?;
 
         self.delete_merged_generation_dirs(&merged_paths).await?;
-        self.pinned_version = None;
+        self.clear_version_pin();
         Ok(true)
     }
 
@@ -939,8 +1121,9 @@ impl StorageBase {
     /// only leaks one directory.
     async fn delete_merged_generation_dirs(&self, merged_paths: &[String]) -> LanceResult<()> {
         let phase = timer_start!();
-        let object_store = self.dataset.object_store(None).await?;
-        let branch_path = self.dataset.branch_location().path.clone();
+        let dataset = self.current_dataset();
+        let object_store = dataset.object_store(None).await?;
+        let branch_path = dataset.branch_location().path.clone();
         for path in merged_paths {
             let gen_dir = branch_path
                 .clone()
@@ -978,11 +1161,12 @@ impl StorageBase {
         &self,
         manifest: &ShardManifest,
     ) -> LanceResult<(HashSet<u64>, Vec<String>, Vec<RecordBatch>, Arc<Schema>)> {
-        let base_uri = self.dataset.uri().trim_end_matches('/').to_string();
+        let dataset = self.current_dataset();
+        let base_uri = dataset.uri().trim_end_matches('/').to_string();
         let mut merged_generations: HashSet<u64> = HashSet::new();
         let mut merged_paths: Vec<String> = Vec::new();
         let mut generation_batches: Vec<(u64, Vec<RecordBatch>)> = Vec::new();
-        let merge_schema: Arc<Schema> = Arc::new(self.dataset.schema().into());
+        let merge_schema: Arc<Schema> = Arc::new(dataset.schema().into());
 
         // Read at most `merge_max_generations` generations per pass.
         //
@@ -1049,7 +1233,7 @@ impl StorageBase {
     /// row per key. `UpdateAll` preserves normal LSM last-write-wins semantics
     /// while also making a retry after an interrupted manifest drain idempotent.
     async fn merge_prepared_batches(
-        &mut self,
+        &self,
         batches: Vec<RecordBatch>,
         merge_schema: Arc<Schema>,
     ) -> LanceResult<()> {
@@ -1057,15 +1241,15 @@ impl StorageBase {
             batches.into_iter().map(Ok::<RecordBatch, ArrowError>),
             merge_schema,
         );
-        let mut builder = MergeInsertBuilder::try_new(
-            Arc::new(self.dataset.clone()),
-            vec![self.key_column.clone()],
-        )?;
-        builder.when_matched(WhenMatched::UpdateAll);
-        let job = builder.try_build()?;
-        let (dataset, _) = job.execute_reader(reader).await?;
-        self.dataset = Arc::unwrap_or_clone(dataset);
-        Ok(())
+        let key_column = self.key_column.clone();
+        self.call_dataset_with(|current| async move {
+            let mut builder = MergeInsertBuilder::try_new(current, vec![key_column])?;
+            builder.when_matched(WhenMatched::UpdateAll);
+            let job = builder.try_build()?;
+            let (dataset, _) = job.execute_reader(reader).await?;
+            Ok((Arc::unwrap_or_clone(dataset), ()))
+        })
+        .await
     }
 
     /// Evolve an older base table to the store's latest additive schema.
@@ -1073,34 +1257,41 @@ impl StorageBase {
     /// Missing nullable columns are added as all-null arrays. Existing unknown
     /// columns, type changes, and missing required columns remain hard errors.
     /// A no-op when the store declared no `latest_schema`.
-    pub async fn ensure_latest_schema(&mut self) -> LanceResult<()> {
+    pub async fn ensure_latest_schema(&self) -> LanceResult<()> {
         let Some(latest_schema) = self.latest_schema.clone() else {
             return Ok(());
         };
-        self.refresh_latest().await?;
+        self.with_exclusive_writer(|| async {
+            self.refresh_latest().await?;
 
-        let base_schema: Arc<Schema> = Arc::new(self.dataset.schema().into());
-        align_batch_to_schema(
-            RecordBatch::new_empty(base_schema.clone()),
-            latest_schema.clone(),
-        )?;
+            let base_schema: Arc<Schema> = Arc::new(self.current_dataset().schema().into());
+            align_batch_to_schema(
+                RecordBatch::new_empty(base_schema.clone()),
+                latest_schema.clone(),
+            )?;
 
-        let missing_fields = latest_schema
-            .fields()
-            .iter()
-            .filter(|field| base_schema.field_with_name(field.name()).is_err())
-            .cloned()
-            .collect::<Vec<_>>();
-        if !missing_fields.is_empty() {
-            self.dataset
-                .add_columns(
-                    NewColumnTransform::AllNulls(Arc::new(Schema::new(missing_fields))),
-                    None,
-                    None,
-                )
+            let missing_fields = latest_schema
+                .fields()
+                .iter()
+                .filter(|field| base_schema.field_with_name(field.name()).is_err())
+                .cloned()
+                .collect::<Vec<_>>();
+            if !missing_fields.is_empty() {
+                self.call_dataset_mut_fn(|mut dataset| async move {
+                    dataset
+                        .add_columns(
+                            NewColumnTransform::AllNulls(Arc::new(Schema::new(missing_fields))),
+                            None,
+                            None,
+                        )
+                        .await?;
+                    Ok(dataset)
+                })
                 .await?;
-        }
-        Ok(())
+            }
+            Ok(())
+        })
+        .await
     }
 
     // ------------------------------------------------- compaction & indexing
@@ -1120,7 +1311,7 @@ impl StorageBase {
     /// to call while other workers are appending or WAL-merging: `Append` vs
     /// `Rewrite` is non-conflicting in Lance's matrix.
     pub async fn compact(
-        &mut self,
+        &self,
         options: Option<CompactionConfig>,
     ) -> LanceResult<CompactionMetrics> {
         let config = options.unwrap_or_default();
@@ -1147,39 +1338,56 @@ impl StorageBase {
             ..Default::default()
         };
 
-        let result = match config.max_source_fragments {
-            Some(max_source_fragments) => {
-                compact_files_incremental(
-                    &mut self.dataset,
-                    lance_options,
-                    max_source_fragments.max(1),
-                )
-                .await
-            }
-            None => compact_files(&mut self.dataset, lance_options, None).await,
-        };
+        // Compact then reload as one exclusive section so a checkout cannot
+        // land between the two publishes. Still publish the local handle even
+        // when compact returns Err (preserves prior behavior: Lance may have
+        // partially updated the in-memory view).
+        self.with_exclusive_writer(|| async {
+            let result = self
+                .call_dataset_with(|current| async move {
+                    let mut dataset = (*current).clone();
+                    let result = match config.max_source_fragments {
+                        Some(max_source_fragments) => {
+                            compact_files_incremental(
+                                &mut dataset,
+                                lance_options,
+                                max_source_fragments.max(1),
+                            )
+                            .await
+                        }
+                        None => compact_files(&mut dataset, lance_options, None).await,
+                    };
+                    Ok((dataset, result))
+                })
+                .await?;
 
-        match result {
-            Ok(metrics) => {
-                // Reload the handle so the caller (and subsequent reads on this
-                // instance) observe the compacted version.
-                self.reload().await?;
-                self.last_compaction = Some(Utc::now());
-                self.total_compactions += 1;
-                self.last_compaction_error = None;
-                info!(
-                    fragments_removed = metrics.fragments_removed,
-                    fragments_added = metrics.fragments_added,
-                    "base-table compaction completed"
-                );
-                Ok(metrics)
-            }
-            Err(e) => {
-                warn!(error = %e, "base-table compaction failed");
-                self.last_compaction_error = Some(e.to_string());
-                Err(e)
+            match result {
+                Ok(metrics) => {
+                    self.reload().await?;
+                    {
+                        let mut state = self.compaction.lock().unwrap_or_else(|e| e.into_inner());
+                        state.last_compaction = Some(Utc::now());
+                        state.total_compactions += 1;
+                        state.last_error = None;
+                    }
+                    info!(
+                        fragments_removed = metrics.fragments_removed,
+                        fragments_added = metrics.fragments_added,
+                        "base-table compaction completed"
+                    );
+                    Ok(metrics)
+                }
+                Err(e) => {
+                    warn!(error = %e, "base-table compaction failed");
+                    {
+                        let mut state = self.compaction.lock().unwrap_or_else(|e| e.into_inner());
+                        state.last_error = Some(e.to_string());
+                    }
+                    Err(e)
+                }
             }
-        }
+        })
+        .await
     }
 
     /// Build a ZoneMap scalar index on the base table's key column.
@@ -1197,20 +1405,26 @@ impl StorageBase {
     /// only ever needs to describe the base table's already-merged fragments —
     /// rows still living in unmerged WAL generations are found by the normal
     /// scan of those generations.
-    pub async fn create_key_zonemap_index(&mut self) -> LanceResult<()> {
+    pub async fn create_key_zonemap_index(&self) -> LanceResult<()> {
         info!(column = %self.key_column, "creating ZoneMap index on key column");
-        self.dataset
-            .create_index_builder(
-                &[self.key_column.as_str()],
-                IndexType::ZoneMap,
-                &ScalarIndexParams::default(),
-            )
-            .name(ID_INDEX_NAME.to_string())
-            .replace(true)
+        let key_column = self.key_column.clone();
+        self.with_exclusive_writer(|| async {
+            self.call_dataset_mut_fn(|mut dataset| async move {
+                dataset
+                    .create_index_builder(
+                        &[key_column.as_str()],
+                        IndexType::ZoneMap,
+                        &ScalarIndexParams::default(),
+                    )
+                    .name(ID_INDEX_NAME.to_string())
+                    .replace(true)
+                    .await?;
+                Ok(dataset)
+            })
             .await?;
-        // Reload the handle so subsequent reads on this instance observe the new
-        // index (mirrors the reload done after `compact`).
-        self.reload().await
+            self.reload().await
+        })
+        .await
     }
 
     /// Whether the base table has accumulated at least `min_fragments`
@@ -1219,7 +1433,7 @@ impl StorageBase {
     /// same config it would pass to [`Self::compact`].
     #[must_use]
     pub fn should_compact(&self, config: &CompactionConfig) -> bool {
-        if self.dataset.count_fragments() < config.min_fragments {
+        if self.current_dataset().count_fragments() < config.min_fragments {
             return false;
         }
         if !config.quiet_hours.is_empty() {
@@ -1236,28 +1450,32 @@ impl StorageBase {
 
     /// Current compaction statistics for the base table.
     ///
-    /// `is_compacting` is always `false`: compaction runs synchronously under
-    /// the caller's `&mut self`, so a stats read cannot observe an in-flight
-    /// compaction on this handle.
+    /// `is_compacting` is always `false`: compaction runs synchronously on this
+    /// handle, so a stats read cannot observe an in-flight compaction here.
     #[must_use]
     pub fn compaction_stats(&self) -> CompactionStats {
+        let state = self.compaction.lock().unwrap_or_else(|e| e.into_inner());
         CompactionStats {
-            total_fragments: self.dataset.count_fragments(),
+            total_fragments: self.current_dataset().count_fragments(),
             is_compacting: false,
-            last_compaction: self.last_compaction,
-            last_error: self.last_compaction_error.clone(),
-            total_compactions: self.total_compactions,
+            last_compaction: state.last_compaction,
+            last_error: state.last_error.clone(),
+            total_compactions: state.total_compactions,
         }
     }
 
     /// Reload the base dataset handle through [`Self::load_with_options`], so
     /// the shared session and storage options are never dropped.
-    pub async fn reload(&mut self) -> LanceResult<()> {
-        let uri = self.dataset.uri().to_string();
-        self.dataset =
-            Self::load_with_options(&uri, self.storage_options.clone(), self.session.clone())
-                .await?;
-        self.pinned_version = None;
+    pub async fn reload(&self) -> LanceResult<()> {
+        let uri = self.uri();
+        let storage_options = self.storage_options.clone();
+        let session = self.session.clone();
+        self.call_dataset_with(move |_current| async move {
+            let dataset = Self::load_with_options(&uri, storage_options, session).await?;
+            Ok((dataset, ()))
+        })
+        .await?;
+        self.clear_version_pin();
         Ok(())
     }
 
@@ -1274,33 +1492,35 @@ impl StorageBase {
     /// `RetryableCommitConflict`. That is benign here — the winner created
     /// exactly the index we wanted — so we reload and treat "index now present"
     /// as success. Any other error propagates.
-    async fn ensure_mem_wal(&mut self) -> LanceResult<()> {
+    async fn ensure_mem_wal(&self) -> LanceResult<()> {
         if self.mem_wal_index_present().await? {
             return Ok(());
         }
-        match self
-            .dataset
-            .initialize_mem_wal()
-            .unsharded()
-            .execute()
-            .await
-        {
-            Ok(()) => Ok(()),
-            Err(err) => {
-                // A concurrent first-writer may have created the index between
-                // our check and our commit. Reload and accept it if so.
-                self.reload().await?;
-                if self.mem_wal_index_present().await? {
-                    Ok(())
-                } else {
-                    Err(err)
+        self.with_exclusive_writer(|| async {
+            let init_result = self
+                .call_dataset_with(|current| async move {
+                    let mut dataset = (*current).clone();
+                    dataset.initialize_mem_wal().unsharded().execute().await?;
+                    Ok((dataset, ()))
+                })
+                .await;
+            match init_result {
+                Ok(()) => Ok(()),
+                Err(err) => {
+                    self.reload().await?;
+                    if self.mem_wal_index_present().await? {
+                        Ok(())
+                    } else {
+                        Err(err)
+                    }
                 }
             }
-        }
+        })
+        .await
     }
 
     async fn mem_wal_index_present(&self) -> LanceResult<bool> {
-        let indices = self.dataset.load_indices().await?;
+        let indices = self.current_dataset().load_indices().await?;
         Ok(indices.iter().any(|i| i.name == MEM_WAL_INDEX_NAME))
     }
 
@@ -1309,14 +1529,15 @@ impl StorageBase {
     pub fn flushed_generation_uri(&self, shard_id: Uuid, path: &str) -> String {
         format!(
             "{}/_mem_wal/{shard_id}/{path}",
-            self.dataset.uri().trim_end_matches('/')
+            self.uri().trim_end_matches('/')
         )
     }
 
     /// Open a flushed generation dataset, inheriting the base dataset's session
     /// and this store's storage options.
     pub async fn open_flushed_dataset(&self, uri: &str) -> LanceResult<Dataset> {
-        let mut builder = DatasetBuilder::from_uri(uri).with_session(self.dataset.session());
+        let mut builder =
+            DatasetBuilder::from_uri(uri).with_session(self.current_dataset().session());
         if let Some(options) = self.storage_options.clone() {
             builder = builder.with_storage_options(options);
         }
@@ -1327,9 +1548,10 @@ impl StorageBase {
     /// bounded-concurrent so stores with many writer instances do not pay one
     /// object-store round trip per shard serially.
     pub async fn wal_shard_snapshots(&self) -> LanceResult<Vec<ShardSnapshot>> {
-        let object_store = self.dataset.object_store(None).await?;
-        let branch_path = self.dataset.branch_location().path.clone();
-        let shard_ids = self.dataset.list_mem_wal_latest_shard_ids().await?;
+        let dataset = self.current_dataset();
+        let object_store = dataset.object_store(None).await?;
+        let branch_path = dataset.branch_location().path.clone();
+        let shard_ids = dataset.list_mem_wal_latest_shard_ids().await?;
 
         let snapshots: Vec<Option<ShardSnapshot>> = stream::iter(shard_ids)
             .map(|shard_id| {
@@ -1385,7 +1607,7 @@ impl StorageBase {
                 })
             })
             .collect();
-        let session = self.dataset.session();
+        let session = self.current_dataset().session();
         let storage_options = self.storage_options.clone();
 
         stream::iter(generation_paths)
@@ -1431,22 +1653,19 @@ impl StorageBase {
         shard_snapshots: Vec<ShardSnapshot>,
     ) -> LsmScanner {
         let merge_key = vec![self.key_column.clone()];
+        let dataset = self.current_dataset();
         match source {
-            ListSource::Fragments => {
-                LsmScanner::new(Arc::new(self.dataset.clone()), Vec::new(), merge_key)
-            }
-            ListSource::All => {
-                LsmScanner::new(Arc::new(self.dataset.clone()), shard_snapshots, merge_key)
-            }
+            ListSource::Fragments => LsmScanner::new(dataset, Vec::new(), merge_key),
+            ListSource::All => LsmScanner::new(dataset, shard_snapshots, merge_key),
             ListSource::Wal => {
-                let arrow_schema: Schema = self.dataset.schema().into();
+                let arrow_schema: Schema = dataset.schema().into();
                 LsmScanner::without_base_table(
                     Arc::new(arrow_schema),
-                    self.dataset.uri().trim_end_matches('/').to_string(),
+                    dataset.uri().trim_end_matches('/').to_string(),
                     shard_snapshots,
                     merge_key,
                 )
-                .with_session(self.dataset.session())
+                .with_session(dataset.session())
             }
         }
     }
@@ -1711,3 +1930,299 @@ pub fn derive_shard_id(instance_id: Option<&str>) -> Uuid {
     let input = instance_id.unwrap_or("default");
     Uuid::new_v5(&Uuid::NAMESPACE_OID, input.as_bytes())
 }
+
+/// Test-only controls for proving ArcSwap handle lost-updates.
+///
+/// Integration tests cannot see `cfg(test)` on this crate, so deterministic
+/// RMW races live in unit tests below that use these hooks.
+#[cfg(test)]
+pub(crate) mod dataset_rmw_test_hooks {
+    use std::sync::{Arc, Mutex};
+
+    use tokio::sync::Notify;
+
+    /// These hooks are process-global; RMW tests that use them must not overlap.
+    static SERIAL: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+    static BEFORE_PUBLISH: Mutex<Option<Arc<Notify>>> = Mutex::new(None);
+    static READY_TO_PUBLISH: Mutex<Option<Arc<Notify>>> = Mutex::new(None);
+
+    pub async fn serial() -> tokio::sync::MutexGuard<'static, ()> {
+        SERIAL.lock().await
+    }
+
+    /// Install a barrier: the next `before_publish` notifies `ready`, then waits
+    /// on `proceed`. Returns `(ready, proceed)`.
+    pub fn install_before_publish_barrier() -> (Arc<Notify>, Arc<Notify>) {
+        let ready = Arc::new(Notify::new());
+        let proceed = Arc::new(Notify::new());
+        *BEFORE_PUBLISH.lock().unwrap() = Some(proceed.clone());
+        *READY_TO_PUBLISH.lock().unwrap() = Some(ready.clone());
+        (ready, proceed)
+    }
+
+    pub fn clear() {
+        *BEFORE_PUBLISH.lock().unwrap() = None;
+        *READY_TO_PUBLISH.lock().unwrap() = None;
+    }
+
+    pub async fn await_before_publish() {
+        let ready = READY_TO_PUBLISH.lock().unwrap().clone();
+        let proceed = BEFORE_PUBLISH.lock().unwrap().clone();
+        let (Some(ready), Some(proceed)) = (ready, proceed) else {
+            return;
+        };
+        // One-shot: clear so only the first publisher hits the barrier.
+        *READY_TO_PUBLISH.lock().unwrap() = None;
+        *BEFORE_PUBLISH.lock().unwrap() = None;
+        ready.notify_one();
+        proceed.notified().await;
+    }
+}
+
+#[cfg(test)]
+mod dataset_handle_rmw_tests {
+    use super::dataset_rmw_test_hooks as hooks;
+    use crate::{RolloutRecord, RolloutStore, RolloutStoreOptions, ROLE_ASSISTANT};
+    use std::sync::Arc;
+
+    fn rec(id: &str) -> RolloutRecord {
+        RolloutRecord {
+            id: id.to_string(),
+            rollout_id: "r".to_string(),
+            problem_id: "p".to_string(),
+            dataset: Some("d".to_string()),
+            sequence_order: 0,
+            role: ROLE_ASSISTANT.to_string(),
+            created_at: chrono::Utc::now(),
+            content: Some("x".to_string()),
+            content_type: "text/plain".to_string(),
+            model_input_string: None,
+            model_output_string: None,
+            rationale: None,
+            problem_text: None,
+            user_metadata: None,
+            input_tokens: None,
+            output_tokens: None,
+            num_input_tokens: None,
+            num_output_tokens: None,
+            output_logprobs: None,
+            input_logprobs: None,
+            ref_logprobs: None,
+            loss_mask: None,
+            advantage: None,
+            reward: None,
+            raw_reward: None,
+            grader_id: None,
+            score: None,
+            include_in_training: None,
+            exclude_reason: None,
+            policy_version: None,
+            relationships: vec![],
+            binary_payload: None,
+            payload_size: None,
+            payload_checksum: None,
+            artifact_type: None,
+            metadata: None,
+        }
+    }
+
+    async fn merge_once(store: &RolloutStore) -> usize {
+        let prepared = store.prepare_cleanup_merge().await.unwrap();
+        match prepared {
+            Some((ms, m, p)) => store.commit_prepared_merge(&ms, &m, p).await.unwrap(),
+            None => 0,
+        }
+    }
+
+    /// Refresh holds `write_writer` across checkout_latest→store, so a concurrent
+    /// merge cannot publish in between. After refresh completes, merge proceeds
+    /// and the in-memory version never goes backwards.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn merge_waits_on_refresh_write_writer() {
+        use std::time::Duration;
+
+        let _serial = hooks::serial().await;
+        hooks::clear();
+        let tmp = tempfile::tempdir().unwrap();
+        let uri = tmp.path().to_string_lossy().to_string();
+        let store = Arc::new(
+            RolloutStore::open_with_options(
+                &uri,
+                RolloutStoreOptions {
+                    shard_id: Some("solo".into()),
+                    merge_after_generations: None,
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap(),
+        );
+
+        for i in 0..4 {
+            store.add(&[rec(&format!("pre-{i}"))]).await.unwrap();
+            store.flush().await.unwrap();
+        }
+        assert!(merge_once(&store).await > 0);
+
+        for i in 0..3 {
+            store.add(&[rec(&format!("pending-{i}"))]).await.unwrap();
+            store.flush().await.unwrap();
+        }
+
+        hooks::clear();
+        let (ready, proceed) = hooks::install_before_publish_barrier();
+
+        let refresher = {
+            let store = store.clone();
+            tokio::spawn(async move {
+                store.refresh_latest().await.unwrap();
+            })
+        };
+
+        ready.notified().await;
+        let v_at_pause = store.version();
+
+        let merger = {
+            let store = store.clone();
+            tokio::spawn(async move { merge_once(&store).await })
+        };
+
+        tokio::time::sleep(Duration::from_millis(150)).await;
+        assert!(
+            !merger.is_finished(),
+            "merge must wait on write_writer held by refresh"
+        );
+
+        proceed.notify_one();
+        refresher.await.unwrap();
+        let reclaimed = merger.await.unwrap();
+        hooks::clear();
+        assert!(reclaimed > 0);
+        assert!(
+            store.version() >= v_at_pause,
+            "handle version must not roll back after refresh then merge"
+        );
+        for i in 0..3 {
+            assert!(store
+                .get_by_id(&format!("pending-{i}"))
+                .await
+                .unwrap()
+                .is_some());
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn merge_after_checkout_clears_pin_and_lands_on_tip() {
+        let _serial = hooks::serial().await;
+        hooks::clear();
+        let tmp = tempfile::tempdir().unwrap();
+        let uri = tmp.path().to_string_lossy().to_string();
+        let store = RolloutStore::open_with_options(
+            &uri,
+            RolloutStoreOptions {
+                shard_id: Some("solo".into()),
+                merge_after_generations: None,
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap();
+
+        for i in 0..3 {
+            store.add(&[rec(&format!("a-{i}"))]).await.unwrap();
+            store.flush().await.unwrap();
+        }
+        assert!(merge_once(&store).await > 0);
+        let pinned = store.version();
+        store.checkout(pinned).await.unwrap();
+        assert!(store.is_version_pinned());
+
+        for i in 0..3 {
+            store.add(&[rec(&format!("b-{i}"))]).await.unwrap();
+            store.flush().await.unwrap();
+        }
+        assert!(merge_once(&store).await > 0);
+        assert!(
+            !store.is_version_pinned(),
+            "rollout merge refresh should clear an explicit checkout pin"
+        );
+        let tip = store.version();
+        store.refresh_latest().await.unwrap();
+        assert_eq!(store.version(), tip, "handle should already be at tip");
+        for i in 0..3 {
+            assert!(store.get_by_id(&format!("b-{i}")).await.unwrap().is_some());
+        }
+    }
+
+    /// A shared AtomicBool "held" flag would let task B see task A's hold and
+    /// skip `write_writer` — defeating the critical section. Task-local nesting
+    /// must make B wait on the mutex instead.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn exclusive_writer_held_is_task_local() {
+        use arrow_schema::{DataType, Field, Schema};
+        use std::time::Duration;
+        use tokio::sync::Notify;
+
+        use super::{StorageBase, StorageBaseOptions};
+
+        let tmp = tempfile::tempdir().unwrap();
+        let uri = tmp.path().to_string_lossy().to_string();
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Utf8, false)]));
+        let base = Arc::new(
+            StorageBase::open(
+                &uri,
+                StorageBaseOptions {
+                    storage_options: None,
+                    shard_id: Some("solo".into()),
+                    merge_after_generations: None,
+                    session: None,
+                    schema,
+                    key_column: "id".into(),
+                    latest_schema: None,
+                    seal_on_put: true,
+                },
+                true,
+            )
+            .await
+            .unwrap(),
+        );
+
+        let ready = Arc::new(Notify::new());
+        let proceed = Arc::new(Notify::new());
+
+        let holder = {
+            let base = Arc::clone(&base);
+            let ready = Arc::clone(&ready);
+            let proceed = Arc::clone(&proceed);
+            tokio::spawn(async move {
+                base.with_exclusive_writer(|| async {
+                    ready.notify_one();
+                    proceed.notified().await;
+                    Ok(())
+                })
+                .await
+                .unwrap();
+            })
+        };
+
+        ready.notified().await;
+
+        let waiter = {
+            let base = Arc::clone(&base);
+            tokio::spawn(async move {
+                base.refresh_latest().await.unwrap();
+            })
+        };
+
+        // If held leaked across tasks, waiter would finish without waiting.
+        tokio::time::sleep(Duration::from_millis(150)).await;
+        assert!(
+            !waiter.is_finished(),
+            "another task must block on write_writer, not skip via a shared held flag"
+        );
+
+        proceed.notify_one();
+        holder.await.unwrap();
+        waiter.await.unwrap();
+    }
+}
diff --git a/crates/lance-context-core/tests/wal_merge_concurrency.rs b/crates/lance-context-core/tests/wal_merge_concurrency.rs
index c2025fe..d0d3c85 100644
--- a/crates/lance-context-core/tests/wal_merge_concurrency.rs
+++ b/crates/lance-context-core/tests/wal_merge_concurrency.rs
@@ -1,3 +1,5 @@
+#![recursion_limit = "256"]
+
 //! Concurrency tests for WAL self-merge: a merge must never block or corrupt
 //! concurrent appends.
 //!
@@ -15,28 +17,38 @@
 //!
 //! 1. appends succeed while a merge runs, and no row is lost;
 //! 2. a generation sealed *during* a merge is not silently dropped by the drain;
-//! 3. concurrent merges do not duplicate rows;
+//! 3. concurrent merges do not duplicate rows, and `merge_lock` excludes a
+//!    second prepare while the first `PreparedMerge` is still live;
 //! 4. an interrupted merge loses nothing (rows stay readable exactly once);
-//! 5. `add` is not blocked for the merge's duration.
-
+//! 5. `add` is not blocked for the merge's duration;
+//! 6. `add` is not blocked for a base-table compact's duration;
+//! 7. concurrent `refresh_latest` cannot roll the in-memory dataset handle
+//!    backwards over a merge's published version (`write_writer` serializes
+//!    handle publish);
+//! 8. same handle monotonicity under concurrent refresh vs compact;
+//! 9. merge and compact can run together without losing rows;
+//! 10. `get_by_id` does not flaky-miss merged rows under a refresh storm.
+
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
-use lance_context_core::{RolloutRecord, RolloutStore, RolloutStoreOptions, ROLE_ASSISTANT};
+use lance_context_core::{
+    CompactionConfig, RolloutRecord, RolloutStore, RolloutStoreOptions, ROLE_ASSISTANT,
+};
 use tokio::sync::RwLock;
 
-/// Run one full merge exactly the way the server's sweepers do: seal + read the
-/// generations under a **read** lock (so appends keep running), then take the
-/// write lock only for the short commit. Returns generations reclaimed.
+/// Run one full merge: seal + read generations, then commit. Both phases use
+/// `&self` on the store (dataset handle is ArcSwap), so callers only need a
+/// shared lock — appends are not blocked. Returns generations reclaimed.
 ///
 /// Every test drives merges through this helper so the lock discipline under
-/// test is the same one production uses -- a test that merged under a single
-/// exclusive lock would pass while the real stall persisted.
+/// test matches production.
 async fn merge_like_sweeper(store: &Arc<RwLock<RolloutStore>>) -> usize {
     let prepared = { store.read().await.prepare_cleanup_merge().await.unwrap() };
     match prepared {
         Some((manifest_store, manifest, prepared)) => store
-            .write()
+            .read()
             .await
             .commit_prepared_merge(&manifest_store, &manifest, prepared)
             .await
@@ -255,8 +267,9 @@ async fn generation_sealed_during_merge_is_not_dropped() {
 
 /// Two merges racing must not append the same generations twice.
 ///
-/// Merges are serialized by an internal mutex (not the store lock, which would
-/// also exclude appends); the loser returns 0 rather than waiting.
+/// Merges are serialized by `StorageBase`'s internal `merge_lock` (prepare
+/// through commit). A `try_lock` loser gets `prepare_* -> None` / reclaim `0`
+/// rather than waiting — appends never take this lock.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn concurrent_merges_do_not_duplicate_rows() {
     let tmp = tempfile::tempdir().unwrap();
@@ -285,11 +298,23 @@ async fn concurrent_merges_do_not_duplicate_rows() {
             async move { merge_like_sweeper(&store).await },
         ));
     }
+    let mut reclaimed = Vec::new();
     for h in handles {
         // None may error; a loser simply reports 0.
-        h.await.unwrap();
+        reclaimed.push(h.await.unwrap());
     }
 
+    let winners: Vec<usize> = reclaimed.iter().copied().filter(|&n| n > 0).collect();
+    assert_eq!(
+        winners.len(),
+        1,
+        "exactly one merge may reclaim; got reclaimed={reclaimed:?}"
+    );
+    assert_eq!(
+        winners[0], 10,
+        "winner should reclaim every pending generation; got reclaimed={reclaimed:?}"
+    );
+
     let ids = read_ids(&store).await;
     let mut deduped = ids.clone();
     deduped.dedup();
@@ -300,6 +325,92 @@ async fn concurrent_merges_do_not_duplicate_rows() {
     assert_eq!(ids.len(), 10, "all rows readable exactly once: {ids:?}");
 }
 
+/// Direct mutual-exclusion check for `merge_lock`: while one `PreparedMerge`
+/// is live (prepare done, commit not yet), a second `prepare_*` must lose
+/// `try_lock` and return `None` — even though both only hold the outer store
+/// `RwLock` for shared/`read` access.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn second_merge_prepare_is_rejected_while_first_holds_prepared_merge() {
+    let tmp = tempfile::tempdir().unwrap();
+    let uri = tmp.path().to_string_lossy().to_string();
+    let store = Arc::new(RwLock::new(
+        RolloutStore::open_with_options(&uri, opts("solo"))
+            .await
+            .unwrap(),
+    ));
+
+    for i in 0..4 {
+        store
+            .read()
+            .await
+            .add(&[rec(&format!("row-{i}"))])
+            .await
+            .unwrap();
+        store.read().await.flush().await.unwrap();
+    }
+
+    let first = {
+        let guard = store.read().await;
+        guard.prepare_cleanup_merge().await.unwrap()
+    };
+    let Some((manifest_store, manifest, prepared)) = first else {
+        panic!("expected pending generations for the first prepare");
+    };
+
+    // Outer store lock is already dropped; only `_merge_guard` inside
+    // `prepared` serializes merges. A concurrent prepare must no-op.
+    let second = {
+        let guard = store.read().await;
+        guard.prepare_cleanup_merge().await.unwrap()
+    };
+    assert!(
+        second.is_none(),
+        "second prepare must lose merge_lock try_lock while PreparedMerge is live"
+    );
+
+    // Appends must still flow under the held merge lock.
+    store
+        .read()
+        .await
+        .add(&[rec("during-held-merge")])
+        .await
+        .unwrap();
+
+    let reclaimed = {
+        let guard = store.read().await;
+        guard
+            .commit_prepared_merge(&manifest_store, &manifest, prepared)
+            .await
+            .unwrap()
+    };
+    assert_eq!(reclaimed, 4);
+
+    // Lock released with PreparedMerge; nothing left to merge until a new seal.
+    let after_commit = {
+        let guard = store.read().await;
+        guard.prepare_cleanup_merge().await.unwrap()
+    };
+    // prepare_cleanup_merge seals first, so the during-held-merge row becomes
+    // one pending generation — that prepare must succeed now that the lock is free.
+    assert!(
+        after_commit.is_some(),
+        "after commit, merge_lock must be free for a new prepare"
+    );
+    let (manifest_store, manifest, prepared) = after_commit.unwrap();
+    let reclaimed = {
+        let guard = store.read().await;
+        guard
+            .commit_prepared_merge(&manifest_store, &manifest, prepared)
+            .await
+            .unwrap()
+    };
+    assert_eq!(reclaimed, 1);
+
+    let ids = read_ids(&store).await;
+    assert_eq!(ids.len(), 5, "all rows readable exactly once: {ids:?}");
+    assert!(ids.contains(&"during-held-merge".to_string()));
+}
+
 /// A merge abandoned partway (the sweeper's timeout does exactly this) must not
 /// lose data. A retry merge-inserts by id, so nothing may disappear or remain
 /// duplicated after a later merge converges.
@@ -414,3 +525,369 @@ async fn append_is_not_blocked_for_the_duration_of_a_merge() {
     );
     assert_eq!(ids.len(), 26, "all rows readable exactly once");
 }
+
+/// Compact is `&self` (ArcSwap dataset handle) and must not take the outer
+/// store write lock — otherwise every concurrent `add` waits for the whole
+/// rewrite. Assert append finishes within a timeout and remains readable; do
+/// not compare wall times to compact (flaky on fast disks).
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn append_is_not_blocked_for_the_duration_of_a_compact() {
+    let tmp = tempfile::tempdir().unwrap();
+    let uri = tmp.path().to_string_lossy().to_string();
+    let store = Arc::new(RwLock::new(
+        RolloutStore::open_with_options(&uri, opts("solo"))
+            .await
+            .unwrap(),
+    ));
+
+    // Many small base-table fragments so compact has real rewrite work.
+    for i in 0..20 {
+        store
+            .read()
+            .await
+            .add(&[rec(&format!("bulk-{i}"))])
+            .await
+            .unwrap();
+        store.read().await.flush().await.unwrap();
+        // Fold each generation into the base table as its own fragment.
+        assert!(merge_like_sweeper(&store).await > 0);
+    }
+
+    let fragments_before = store.read().await.observe().await.unwrap().fragment_count;
+    assert!(
+        fragments_before > 1,
+        "need several base fragments to compact, got {fragments_before}"
+    );
+
+    let compactor = {
+        let store = store.clone();
+        tokio::spawn(async move {
+            let guard = store.read().await;
+            guard
+                .compact(Some(CompactionConfig {
+                    min_fragments: 2,
+                    num_threads: Some(1),
+                    batch_size: Some(1),
+                    ..Default::default()
+                }))
+                .await
+                .unwrap()
+        })
+    };
+
+    // Give compact a moment to enter the rewrite.
+    tokio::time::sleep(Duration::from_millis(10)).await;
+
+    // Fail only on multi-second stalls — not on wall-clock ratios vs compact.
+    const APPEND_NOT_STALLED: Duration = Duration::from_secs(5);
+    tokio::time::timeout(APPEND_NOT_STALLED, async {
+        store
+            .read()
+            .await
+            .add(&[rec("during-compact")])
+            .await
+            .unwrap();
+    })
+    .await
+    .expect("append during compact stalled; it appears blocked behind compact");
+
+    let metrics = compactor.await.unwrap();
+    assert!(
+        metrics.fragments_removed > 0,
+        "compact should have rewritten fragments"
+    );
+
+    store.read().await.flush().await.unwrap();
+    let ids = read_ids(&store).await;
+    assert!(
+        ids.contains(&"during-compact".to_string()),
+        "the row appended during compact must be readable"
+    );
+    assert_eq!(ids.len(), 21, "all rows readable exactly once: {ids:?}");
+}
+
+/// Blind `ArcSwap::store` after load→modify→await lets `refresh_latest` publish
+/// an older handle over a concurrent merge append. That rolls the in-memory
+/// version backwards: object storage still has the merge, but `get_by_id` on
+/// the base handle can flaky-miss until the next refresh (#234 class).
+///
+/// Exclusive `write_writer` serializes handle publish so sampled versions never
+/// decrease and merged rows stay visible immediately after commit.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn refresh_cannot_roll_back_dataset_handle_over_merge() {
+    let tmp = tempfile::tempdir().unwrap();
+    let uri = tmp.path().to_string_lossy().to_string();
+    let store = Arc::new(RwLock::new(
+        RolloutStore::open_with_options(&uri, opts("solo"))
+            .await
+            .unwrap(),
+    ));
+
+    for i in 0..12 {
+        store
+            .read()
+            .await
+            .add(&[rec(&format!("row-{i}"))])
+            .await
+            .unwrap();
+        store.read().await.flush().await.unwrap();
+    }
+
+    let stop = Arc::new(AtomicBool::new(false));
+    let max_seen = Arc::new(AtomicU64::new(store.read().await.version()));
+    let dips = Arc::new(AtomicU64::new(0));
+    let refresher = {
+        let store = store.clone();
+        let stop = stop.clone();
+        let max_seen = max_seen.clone();
+        let dips = dips.clone();
+        tokio::spawn(async move {
+            while !stop.load(Ordering::Acquire) {
+                store.read().await.refresh_latest().await.unwrap();
+                let v = store.read().await.version();
+                let prev_max = max_seen.fetch_max(v, Ordering::SeqCst);
+                if v < prev_max {
+                    dips.fetch_add(1, Ordering::SeqCst);
+                }
+            }
+        })
+    };
+
+    let mut extra = 0usize;
+    for _ in 0..3 {
+        for _ in 0..4 {
+            store
+                .read()
+                .await
+                .add(&[rec(&format!("extra-{extra}"))])
+                .await
+                .unwrap();
+            store.read().await.flush().await.unwrap();
+            extra += 1;
+        }
+        let reclaimed = merge_like_sweeper(&store).await;
+        assert!(reclaimed > 0, "merge should reclaim pending generations");
+        let v = store.read().await.version();
+        let prev_max = max_seen.fetch_max(v, Ordering::SeqCst);
+        assert!(
+            v >= prev_max,
+            "merge published version {v} below previously seen max {prev_max}"
+        );
+    }
+
+    stop.store(true, Ordering::Release);
+    refresher.await.unwrap();
+
+    assert_eq!(
+        dips.load(Ordering::SeqCst),
+        0,
+        "in-memory dataset version went backwards under concurrent refresh_latest"
+    );
+
+    let ids = read_ids(&store).await;
+    for i in 0..12 {
+        assert!(
+            ids.contains(&format!("row-{i}")),
+            "merged row-{i} missing from handle after refresh race: {ids:?}"
+        );
+    }
+}
+
+async fn track_version_dips(
+    store: Arc<RwLock<RolloutStore>>,
+    stop: Arc<AtomicBool>,
+    max_seen: Arc<AtomicU64>,
+    dips: Arc<AtomicU64>,
+) {
+    while !stop.load(Ordering::Acquire) {
+        // Compact can briefly make a concurrent checkout_latest miss a
+        // mid-rewrite manifest; retry rather than failing the storm.
+        if store.read().await.refresh_latest().await.is_err() {
+            continue;
+        }
+        let v = store.read().await.version();
+        let prev_max = max_seen.fetch_max(v, Ordering::SeqCst);
+        if v < prev_max {
+            dips.fetch_add(1, Ordering::SeqCst);
+        }
+    }
+}
+
+/// Compact publishes a new handle (then reloads); concurrent refresh must not
+/// roll the in-memory version backwards over that publish.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn refresh_cannot_roll_back_dataset_handle_over_compact() {
+    let tmp = tempfile::tempdir().unwrap();
+    let uri = tmp.path().to_string_lossy().to_string();
+    let store = Arc::new(RwLock::new(
+        RolloutStore::open_with_options(&uri, opts("solo"))
+            .await
+            .unwrap(),
+    ));
+
+    for i in 0..12 {
+        store
+            .read()
+            .await
+            .add(&[rec(&format!("row-{i}"))])
+            .await
+            .unwrap();
+        store.read().await.flush().await.unwrap();
+        assert!(merge_like_sweeper(&store).await > 0);
+    }
+
+    let stop = Arc::new(AtomicBool::new(false));
+    let max_seen = Arc::new(AtomicU64::new(store.read().await.version()));
+    let dips = Arc::new(AtomicU64::new(0));
+    let refresher = tokio::spawn(track_version_dips(
+        store.clone(),
+        stop.clone(),
+        max_seen.clone(),
+        dips.clone(),
+    ));
+
+    let metrics = {
+        let guard = store.read().await;
+        guard
+            .compact(Some(CompactionConfig {
+                min_fragments: 2,
+                num_threads: Some(1),
+                batch_size: Some(1),
+                ..Default::default()
+            }))
+            .await
+            .unwrap()
+    };
+    assert!(
+        metrics.fragments_removed > 0,
+        "compact should rewrite fragments, got removed={} added={}",
+        metrics.fragments_removed,
+        metrics.fragments_added
+    );
+    let v = store.read().await.version();
+    let prev_max = max_seen.fetch_max(v, Ordering::SeqCst);
+    assert!(
+        v >= prev_max,
+        "compact published version {v} below previously seen max {prev_max}"
+    );
+
+    stop.store(true, Ordering::Release);
+    refresher.await.unwrap();
+    assert_eq!(
+        dips.load(Ordering::SeqCst),
+        0,
+        "in-memory dataset version went backwards under concurrent refresh vs compact"
+    );
+    assert_eq!(read_ids(&store).await.len(), 12);
+}
+
+/// Lance treats Append (WAL merge) vs Rewrite (compact) as non-conflicting;
+/// both must succeed and conserve rows under shared store locks.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn merge_and_compact_concurrently_preserve_rows() {
+    let tmp = tempfile::tempdir().unwrap();
+    let uri = tmp.path().to_string_lossy().to_string();
+    let store = Arc::new(RwLock::new(
+        RolloutStore::open_with_options(&uri, opts("solo"))
+            .await
+            .unwrap(),
+    ));
+
+    for i in 0..8 {
+        store
+            .read()
+            .await
+            .add(&[rec(&format!("base-{i}"))])
+            .await
+            .unwrap();
+        store.read().await.flush().await.unwrap();
+        assert!(merge_like_sweeper(&store).await > 0);
+    }
+    for i in 0..6 {
+        store
+            .read()
+            .await
+            .add(&[rec(&format!("wal-{i}"))])
+            .await
+            .unwrap();
+        store.read().await.flush().await.unwrap();
+    }
+
+    let merger = {
+        let store = store.clone();
+        tokio::spawn(async move { merge_like_sweeper(&store).await })
+    };
+    let compactor = {
+        let store = store.clone();
+        tokio::spawn(async move {
+            store
+                .read()
+                .await
+                .compact(Some(CompactionConfig {
+                    min_fragments: 2,
+                    num_threads: Some(1),
+                    ..Default::default()
+                }))
+                .await
+                .unwrap()
+        })
+    };
+
+    let reclaimed = merger.await.unwrap();
+    let metrics = compactor.await.unwrap();
+    assert!(reclaimed > 0, "merge should reclaim WAL generations");
+    let _ = metrics;
+
+    let ids = read_ids(&store).await;
+    assert_eq!(ids.len(), 14, "all rows readable exactly once: {ids:?}");
+    for i in 0..8 {
+        assert!(ids.contains(&format!("base-{i}")));
+    }
+    for i in 0..6 {
+        assert!(ids.contains(&format!("wal-{i}")));
+    }
+}
+
+/// Production symptom of handle rollback: merged id briefly missing from
+/// `get_by_id` while refresh races merge. Exclusive handle publish keeps the
+/// id visible once merge returns.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn get_by_id_sees_merged_rows_under_refresh_storm() {
+    let tmp = tempfile::tempdir().unwrap();
+    let uri = tmp.path().to_string_lossy().to_string();
+    let store = Arc::new(RwLock::new(
+        RolloutStore::open_with_options(&uri, opts("solo"))
+            .await
+            .unwrap(),
+    ));
+
+    let stop = Arc::new(AtomicBool::new(false));
+    let refresher = {
+        let store = store.clone();
+        let stop = stop.clone();
+        tokio::spawn(async move {
+            while !stop.load(Ordering::Acquire) {
+                store.read().await.refresh_latest().await.unwrap();
+            }
+        })
+    };
+
+    store.read().await.add(&[rec("target")]).await.unwrap();
+    store.read().await.flush().await.unwrap();
+    assert!(merge_like_sweeper(&store).await > 0);
+
+    for _ in 0..30 {
+        let hit = store
+            .read()
+            .await
+            .get_by_id("target")
+            .await
+            .unwrap()
+            .is_some();
+        assert!(hit, "merged target must remain visible under refresh storm");
+        tokio::time::sleep(Duration::from_millis(2)).await;
+    }
+
+    stop.store(true, Ordering::Release);
+    refresher.await.unwrap();
+}
diff --git a/crates/lance-context-core/tests/wal_merge_generation_cleanup.rs b/crates/lance-context-core/tests/wal_merge_generation_cleanup.rs
index 0848448..ae8e505 100644
--- a/crates/lance-context-core/tests/wal_merge_generation_cleanup.rs
+++ b/crates/lance-context-core/tests/wal_merge_generation_cleanup.rs
@@ -83,7 +83,7 @@ async fn serial_merge_deletes_merged_generation_dirs() {
         ..Default::default()
     };
 
-    let mut store = RolloutStore::open_with_options(&uri, opts.clone())
+    let store = RolloutStore::open_with_options(&uri, opts.clone())
         .await
         .unwrap();
 
diff --git a/crates/lance-context-metrics/src/lib.rs b/crates/lance-context-metrics/src/lib.rs
index 3c54d9c..c0e6c52 100644
--- a/crates/lance-context-metrics/src/lib.rs
+++ b/crates/lance-context-metrics/src/lib.rs
@@ -175,12 +175,12 @@ fn describe_metrics() {
     describe_histogram!(
         "rollout_wal_merge_lock_wait_seconds",
         Unit::Seconds,
-        "Time waiting for the store write lock before a WAL merge (blocks all ingest)."
+        "Time waiting for the store shared lock before a WAL merge prepare."
     );
     describe_histogram!(
         "rollout_compaction_lock_wait_seconds",
         Unit::Seconds,
-        "Time waiting for the store write lock before compaction."
+        "Time waiting for the store shared lock before compaction."
     );
 
     // Master task lifecycle.
diff --git a/crates/lance-context-server/src/routes/compact.rs b/crates/lance-context-server/src/routes/compact.rs
index 0c2c4a9..d8f6902 100644
--- a/crates/lance-context-server/src/routes/compact.rs
+++ b/crates/lance-context-server/src/routes/compact.rs
@@ -28,7 +28,7 @@ pub async fn compact(
         None
     };
 
-    let mut store = store_lock.write().await;
+    let store = store_lock.read().await;
     let metrics = store.compact(config).await.map_err(AppError::from_lance)?;
 
     Ok(Json(CompactResponse {
diff --git a/crates/lance-context-server/src/routes/datagen.rs b/crates/lance-context-server/src/routes/datagen.rs
index 98ec618..fdb1dda 100644
--- a/crates/lance-context-server/src/routes/datagen.rs
+++ b/crates/lance-context-server/src/routes/datagen.rs
@@ -174,7 +174,7 @@ async fn fold_datagen_item_refreshing_on_miss(
         }
     }
 
-    let mut store = store_lock.write().await;
+    let store = store_lock.read().await;
     if !store.is_version_pinned() {
         store.refresh_latest().await.map_err(AppError::from_lance)?;
     }
@@ -197,7 +197,7 @@ async fn datagen_failures_refreshing_on_empty(
         }
     }
 
-    let mut store = store_lock.write().await;
+    let store = store_lock.read().await;
     if !store.is_version_pinned() {
         store.refresh_latest().await.map_err(AppError::from_lance)?;
     }
@@ -220,7 +220,7 @@ async fn datagen_events_for_root_refreshing_on_empty(
         }
     }
 
-    let mut store = store_lock.write().await;
+    let store = store_lock.read().await;
     if !store.is_version_pinned() {
         store.refresh_latest().await.map_err(AppError::from_lance)?;
     }
@@ -243,7 +243,7 @@ async fn datagen_root_statuses_refreshing_on_missing(
         }
     }
 
-    let mut store = store_lock.write().await;
+    let store = store_lock.read().await;
     if !store.is_version_pinned() {
         store.refresh_latest().await.map_err(AppError::from_lance)?;
     }
@@ -266,7 +266,7 @@ async fn get_datagen_blob_refreshing_on_miss(
         }
     }
 
-    let mut store = store_lock.write().await;
+    let store = store_lock.read().await;
     if !store.is_version_pinned() {
         store.refresh_latest().await.map_err(AppError::from_lance)?;
     }
diff --git a/crates/lance-context-server/src/routes/generic.rs b/crates/lance-context-server/src/routes/generic.rs
index 87477a1..2ab0f75 100644
--- a/crates/lance-context-server/src/routes/generic.rs
+++ b/crates/lance-context-server/src/routes/generic.rs
@@ -215,7 +215,7 @@ async fn get_generic_row_refreshing_on_miss(
         }
     }
 
-    let mut store = store_lock.write().await;
+    let store = store_lock.read().await;
     if !store.is_version_pinned() {
         store.refresh_latest().await.map_err(AppError::from_lance)?;
     }
diff --git a/crates/lance-context-server/src/routes/records.rs b/crates/lance-context-server/src/routes/records.rs
index 2d21ae3..0a43441 100644
--- a/crates/lance-context-server/src/routes/records.rs
+++ b/crates/lance-context-server/src/routes/records.rs
@@ -32,7 +32,7 @@ async fn get_context_record_refreshing_on_miss(
         }
     }
 
-    let mut store = store_lock.write().await;
+    let store = store_lock.read().await;
     if !store.is_version_pinned() {
         store.refresh_latest().await.map_err(AppError::from_lance)?;
     }
@@ -73,7 +73,7 @@ async fn fetch_context_payload_refreshing_on_miss(
         }
     }
 
-    let mut store = store_lock.write().await;
+    let store = store_lock.read().await;
     if !store.is_version_pinned() {
         store.refresh_latest().await.map_err(AppError::from_lance)?;
     }
@@ -96,7 +96,7 @@ async fn get_context_by_external_id_refreshing_on_miss(
         }
     }
 
-    let mut store = store_lock.write().await;
+    let store = store_lock.read().await;
     if !store.is_version_pinned() {
         store.refresh_latest().await.map_err(AppError::from_lance)?;
     }
diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs
index f9c7c5a..2ca7e9c 100644
--- a/crates/lance-context-server/src/routes/rollouts.rs
+++ b/crates/lance-context-server/src/routes/rollouts.rs
@@ -307,7 +307,7 @@ pub async fn add_rollouts(
 
     // A read lock: `add` is `&self` and MemWAL appends are internally
     // concurrent, so multiple ingest requests to the same store run in parallel.
-    // Mutating ops (merge, compact, checkout, close) still take the write lock.
+    // Merge exclusivity is `StorageBase::merge_lock`, not this outer write lock.
     let store = store_lock.read().await;
 
     // Times only the store work (`add` + optional `flush`), excluding body
@@ -494,7 +494,7 @@ async fn get_rollout_refreshing_on_miss(
         }
     }
 
-    let mut store = store_lock.write().await;
+    let store = store_lock.read().await;
     if !store.is_version_pinned() {
         store.refresh_latest().await.map_err(AppError::from_lance)?;
     }
@@ -516,7 +516,7 @@ async fn get_rollout_blob_refreshing_on_miss(
         }
     }
 
-    let mut store = store_lock.write().await;
+    let store = store_lock.read().await;
     if !store.is_version_pinned() {
         store.refresh_latest().await.map_err(AppError::from_lance)?;
     }
@@ -575,7 +575,7 @@ pub async fn checkout_rollout(
 ) -> Result<Json<VersionResponse>, AppError> {
     let store_lock = state.get_or_open_rollout_store(&name).await?;
 
-    let mut store = store_lock.write().await;
+    let store = store_lock.read().await;
     store
         .checkout(req.version)
         .await
@@ -613,7 +613,7 @@ pub async fn compact_rollout(
     };
 
     let lock_start = std::time::Instant::now();
-    let mut store = store_lock.write().await;
+    let store = store_lock.read().await;
     ::metrics::histogram!("rollout_compaction_lock_wait_seconds")
         .record(lock_start.elapsed().as_secs_f64());
     let compact_start = std::time::Instant::now();
@@ -671,10 +671,11 @@ pub async fn merge_wal(
     Path(name): Path<String>,
 ) -> Result<Json<MergeWalResponse>, AppError> {
     let store_lock = state.get_or_open_rollout_store(&name).await?;
-    // Split by lock scope: seal + read every flushed generation under the
-    // *read* lock so ingest on this store keeps running, then take the write
-    // lock only for the short commit. Holding the write lock across the whole
-    // merge is what used to stall every concurrent append for its duration.
+    // Store `RwLock` is shared (read) only so we can call `&self` APIs while
+    // ingest keeps running. Merge exclusivity is `StorageBase::merge_lock`
+    // inside prepare→commit (`try_lock`; a loser returns `reclaimed: 0`), not
+    // this outer write lock — holding write across the whole merge is what used
+    // to stall every concurrent append.
     let lock_start = std::time::Instant::now();
     let prepared = {
         let store = store_lock.read().await;
@@ -692,7 +693,7 @@ pub async fn merge_wal(
     let merge_start = std::time::Instant::now();
     let reclaimed = match prepared {
         Some((manifest_store, manifest, prepared)) => {
-            let mut store = store_lock.write().await;
+            let store = store_lock.read().await;
             match store
                 .commit_prepared_merge(&manifest_store, &manifest, prepared)
                 .await
diff --git a/crates/lance-context-server/src/routes/versions.rs b/crates/lance-context-server/src/routes/versions.rs
index 7bb06d2..f99e303 100644
--- a/crates/lance-context-server/src/routes/versions.rs
+++ b/crates/lance-context-server/src/routes/versions.rs
@@ -26,7 +26,7 @@ pub async fn checkout(
 ) -> Result<Json<VersionResponse>, AppError> {
     let store_lock = state.get_or_open_context_store(&name).await?;
 
-    let mut store = store_lock.write().await;
+    let store = store_lock.read().await;
     store
         .checkout(req.version)
         .await
diff --git a/crates/lance-context-server/src/state.rs b/crates/lance-context-server/src/state.rs
index dce945e..ea4e2ea 100644
--- a/crates/lance-context-server/src/state.rs
+++ b/crates/lance-context-server/src/state.rs
@@ -787,7 +787,7 @@ impl AppState {
                 .collect()
         };
         for (name, store) in resident {
-            if let Err(e) = store.write().await.close().await {
+            if let Err(e) = store.read().await.close().await {
                 tracing::warn!(
                     store = %name,
                     error = %e,
@@ -808,7 +808,7 @@ impl AppState {
                 .collect()
         };
         for (name, store) in datagen {
-            if let Err(e) = store.write().await.close().await {
+            if let Err(e) = store.read().await.close().await {
                 tracing::warn!(
                     store = %name,
                     error = %e,
@@ -825,7 +825,7 @@ impl AppState {
                 .collect()
         };
         for (name, store) in generic {
-            if let Err(e) = store.write().await.close().await {
+            if let Err(e) = store.read().await.close().await {
                 tracing::warn!(
                     store = %name,
                     error = %e,
@@ -994,7 +994,7 @@ mod tests {
 
         // --- datagen: seals on append, so the merge pass is what it needs ---
         let datagen_uri = state.datagen_uri("d1");
-        let mut datagen = DatagenStore::open(&datagen_uri).await.unwrap();
+        let datagen = DatagenStore::open(&datagen_uri).await.unwrap();
         datagen.append(&[datagen_event()]).await.unwrap();
         assert!(
             datagen.pending_wal_generations().await.unwrap() > 0,
@@ -1069,6 +1069,6 @@ mod tests {
 
         // The handle survives shutdown (shutdown only drains the writer); a
         // fresh close is still a no-op.
-        store.write().await.close().await.unwrap();
+        store.read().await.close().await.unwrap();
     }
 }
diff --git a/crates/lance-context-server/src/sweeper.rs b/crates/lance-context-server/src/sweeper.rs
index 0500e25..183ac75 100644
--- a/crates/lance-context-server/src/sweeper.rs
+++ b/crates/lance-context-server/src/sweeper.rs
@@ -25,11 +25,9 @@ use tokio::sync::{Mutex, RwLock};
 /// A store the sweepers can maintain.
 ///
 /// Implemented on `Arc<RwLock<Store>>` rather than on the store itself so each
-/// kind decides its own locking. That is load-bearing for rollout, whose merge
-/// deliberately splits into a shared-lock prepare (the expensive object-storage
-/// reads, during which appends keep flowing) and a brief exclusive-lock commit.
-/// A trait over `&mut Store` would have forced the exclusive lock across the
-/// whole merge and quietly stalled the write path.
+/// kind decides its own locking. Merge/flush/commit are `&self` on the store
+/// (dataset handle is ArcSwap; merge exclusivity is an internal `try_lock`), so
+/// these impls only need a shared store lock — concurrent appends keep flowing.
 pub(crate) trait Sweepable: Send + Sync + 'static {
     /// Human-readable kind, for log and metric labels.
     fn kind() -> &'static str;
@@ -48,14 +46,13 @@ impl Sweepable for Arc<RwLock<RolloutStore>> {
     }
 
     async fn flush(&self) -> Result<(), String> {
-        // Read lock: `flush` is `&self`, so concurrent appends are not blocked.
         let guard = self.read().await;
         let result = guard.flush().await.map_err(|e| e.to_string());
         if result.is_ok() {
             // The count-triggered merge rides this timer; it is a no-op unless
             // the threshold is configured and met.
             drop(guard);
-            let mut guard = self.write().await;
+            let guard = self.read().await;
             guard
                 .maybe_merge_own_shard()
                 .await
@@ -65,9 +62,6 @@ impl Sweepable for Arc<RwLock<RolloutStore>> {
     }
 
     async fn merge_wal(&self) -> Result<usize, String> {
-        // The prepare/commit split: seal and read every flushed generation
-        // under the *shared* lock so appends keep running, then take the
-        // exclusive lock only for the short commit.
         let prepared = {
             let guard = self.read().await;
             guard
@@ -78,7 +72,7 @@ impl Sweepable for Arc<RwLock<RolloutStore>> {
         let Some((manifest_store, manifest, prepared)) = prepared else {
             return Ok(0);
         };
-        let mut guard = self.write().await;
+        let guard = self.read().await;
         guard
             .commit_prepared_merge(&manifest_store, &manifest, prepared)
             .await
@@ -99,7 +93,7 @@ impl Sweepable for Arc<RwLock<DatagenStore>> {
     }
 
     async fn merge_wal(&self) -> Result<usize, String> {
-        let mut guard = self.write().await;
+        let guard = self.read().await;
         guard.cleanup_own_shard().await.map_err(|e| e.to_string())
     }
 }
@@ -115,7 +109,7 @@ impl Sweepable for Arc<RwLock<GenericStore>> {
     }
 
     async fn merge_wal(&self) -> Result<usize, String> {
-        let mut guard = self.write().await;
+        let guard = self.read().await;
         guard.cleanup_wal().await.map_err(|e| e.to_string())
     }
 }
diff --git a/python/src/lib.rs b/python/src/lib.rs
index 63504d1..d0c862d 100644
--- a/python/src/lib.rs
+++ b/python/src/lib.rs
@@ -790,7 +790,7 @@ impl Context {
         // two writers racing for one shard. A fork branches the in-memory
         // `Context` and shares the underlying dataset, which a fresh handle
         // gives it.
-        let uri = self.store.uri().to_string();
+        let uri = self.store.uri();
         let store = py.allow_threads(|| self.runtime.block_on(ContextStore::open(&uri)));
         Ok(Self {
             inner: self.inner.fork(branch_name),