refactor(core): migrate DatagenStore onto StorageBase

crates/lance-context-core/src/api_impl.rs

@@ -41,7 +41,10 @@ impl ContextStoreApi for ContextStore {
         }
 
         let count = core_records.len();
-        let version = self.add(&core_records).await.map_err(to_ctx_err)?;
+        // Disambiguate: the inherent `ContextStore::add`, not this trait method.
+        let version = ContextStore::add(self, &core_records)
+            .await
+            .map_err(to_ctx_err)?;
         Ok(AddRecordsResponse {
             version,
             ids,

crates/lance-context-core/src/eval.rs

@@ -529,7 +529,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();
             let a = embedding(&store, &[1.0]);
             let b = embedding(&store, &[0.5]);
             let c = embedding(&store, &[0.0, 1.0]);
@@ -579,7 +579,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();
             let q = embedding(&store, &[1.0]);
             // the only relevant doc is retired -> hidden by default.
             let mut retired = record("doc-a", "alpha", q.clone());
@@ -624,7 +624,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();
             let shared = embedding(&store, &[1.0]);
             let mut a = record("doc-a", "alpha", shared.clone());
             a.tenant = Some("x".to_string());
@@ -662,7 +662,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();
             let a = embedding(&store, &[1.0]);
             let b = embedding(&store, &[0.0, 1.0]);
             store
@@ -701,7 +701,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();
             let a = embedding(&store, &[1.0]);
             let b = embedding(&store, &[0.5]);
             store

crates/lance-context-core/src/namespace.rs

@@ -493,7 +493,7 @@ mod tests {
         let selector = selector(&[("tenant", "acme"), ("source", "memory")]);
         let partition = namespace.resolve_partition(&selector).unwrap();
 
-        let mut store = namespace.context(&selector).await.unwrap();
+        let store = namespace.context(&selector).await.unwrap();
         store
             .add(&[record("rec-1", "acme", "memory")])
             .await

crates/lance-context-core/src/rollout_store.rs

@@ -424,6 +424,11 @@ impl RolloutStore {
                 // claim-check columns), so a WAL merge first evolves an older
                 // base table to the current schema before appending.
                 latest_schema: Some(Arc::new(rollout_schema())),
+                // Rollout defers the seal: high fan-in appends must not
+                // serialize behind a per-append seal, and rollout rows are
+                // immutable so nothing reads back before writing. The server's
+                // flush sweeper (and `?flush=true`) provide visibility.
+                seal_on_put: false,
             },
             create_if_missing,
         )