feat(rollout): MemWAL ingest with server-id sharding

crates/lance-context-api/src/lib.rs

@@ -574,8 +574,6 @@ pub struct CreateRolloutStoreRequest {
     pub name: String,
     #[serde(default)]
     pub storage_options: Option<std::collections::HashMap<String, String>>,
-    #[serde(default)]
-    pub blob_columns: Option<Vec<String>>,
 }
 
 #[derive(Debug, Serialize, Deserialize)]

crates/lance-context-server/Cargo.toml

@@ -17,7 +17,7 @@ lance-context-core = { version = "0.5.1", path = "../lance-context-core" }
 lance-context-api = { version = "0.5.1", path = "../lance-context-api" }
 axum = { version = "0.8", features = ["json", "multipart"] }
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
-clap = { version = "4", features = ["derive"] }
+clap = { version = "4", features = ["derive", "env"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "signal"] }

crates/lance-context-server/src/config.rs

@@ -12,4 +12,28 @@ pub struct ServerConfig {
 
     #[arg(long, default_value = "./lance-data")]
     pub data_dir: String,
+
+    /// Stable identity of this server instance, used as the MemWAL shard key for
+    /// rollout writes. Each instance must present a distinct, stable value so it
+    /// owns exactly one shard and never contends with peers (see
+    /// `specs/rollout-deployment.md`). In Kubernetes, set this to the
+    /// StatefulSet pod ordinal hostname (e.g. `rollout-0`). Defaults to the
+    /// `INSTANCE_ID` env var, then the pod/host `HOSTNAME`; if neither is set,
+    /// rollout writes fall back to a single shared `default` shard, which is
+    /// only safe for a single-instance deployment.
+    #[arg(long, env = "INSTANCE_ID")]
+    pub instance_id: Option<String>,
+}
+
+impl ServerConfig {
+    /// Resolve the instance id used for rollout MemWAL sharding: the explicit
+    /// `--instance-id`/`INSTANCE_ID` if provided, otherwise the `HOSTNAME`
+    /// environment variable (stable per-pod under a StatefulSet).
+    #[must_use]
+    pub fn resolved_instance_id(&self) -> Option<String> {
+        self.instance_id
+            .clone()
+            .or_else(|| std::env::var("HOSTNAME").ok())
+            .filter(|value| !value.is_empty())
+    }
 }

crates/lance-context-server/src/routes/records.rs

@@ -624,6 +624,7 @@ mod tests {
             stores: RwLock::new(stores),
             rollout_stores: RwLock::new(HashMap::new()),
             base_path: dir.path().to_path_buf(),
+            instance_id: None,
         });
         (state, dir)
     }

crates/lance-context-server/src/routes/rollouts.rs

@@ -1,4 +1,4 @@
-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use axum::body::Body;
@@ -36,18 +36,10 @@ pub async fn create_rollout_store(
     }
     drop(stores);
 
-    // `None` keeps the default blob offload of `binary_payload`; an explicit
-    // empty list stores it inline.
-    let blob_columns: HashSet<String> = req
-        .blob_columns
-        .unwrap_or_else(|| vec!["binary_payload".to_string()])
-        .into_iter()
-        .collect();
-
     let uri = state.rollout_uri(&req.name);
     let options = RolloutStoreOptions {
         storage_options: req.storage_options,
-        blob_columns,
+        shard_id: state.instance_id.clone(),
     };
 
     let store = RolloutStore::open_with_options(&uri, options)
@@ -341,7 +333,7 @@ pub async fn get_rollout(
     }))
 }
 
-/// Materialize a rollout row's offloaded `binary_payload` bytes. The artifact
+/// Materialize a rollout row's `binary_payload` bytes. The artifact
 /// bytes are opaque, so they stream as `application/octet-stream`. `404` when
 /// the row is absent or carries no payload.
 pub async fn fetch_rollout_blob(
@@ -512,13 +504,13 @@ mod tests {
             stores: RwLock::new(HashMap::new()),
             rollout_stores: RwLock::new(HashMap::new()),
             base_path: dir.path().to_path_buf(),
+            instance_id: None,
         });
         let (_status, _info) = create_rollout_store(
             State(state.clone()),
             Json(CreateRolloutStoreRequest {
                 name: "rl".to_string(),
                 storage_options: None,
-                blob_columns: None,
             }),
         )
         .await
@@ -575,7 +567,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn add_rollouts_json_inlines_and_offloads_blob() {
+    async fn add_rollouts_json_projects_blob_out_but_fetch_materializes() {
         let (state, _dir) = rollout_state().await;
         let payload = b"artifact-bytes".to_vec();
         let mut record = record_with_size("r0", Some(payload.len() as i64));
@@ -596,7 +588,7 @@ mod tests {
         assert_eq!(resp.count, 1);
         assert_eq!(resp.ids, vec!["r0".to_string()]);
 
-        // A plain get reads the offloaded column back as None...
+        // A plain get projects the binary column out, reading it back as None...
         let Json(got) = get_rollout(
             State(state.clone()),
             Path(("rl".to_string(), "r0".to_string())),

crates/lance-context-server/src/routes/search.rs

@@ -157,6 +157,7 @@ mod tests {
             stores: RwLock::new(stores),
             rollout_stores: RwLock::new(HashMap::new()),
             base_path: dir.path().to_path_buf(),
+            instance_id: None,
         });
         (state, dir)
     }

crates/lance-context-server/src/state.rs

@@ -11,14 +11,20 @@ pub struct AppState {
     pub stores: RwLock<HashMap<String, Arc<RwLock<ContextStore>>>>,
     pub rollout_stores: RwLock<HashMap<String, Arc<RwLock<RolloutStore>>>>,
     pub base_path: PathBuf,
+    /// Stable identity of this server instance, used as the MemWAL shard key for
+    /// rollout writes so each instance owns exactly one shard. `None` falls back
+    /// to a single shared shard (single-instance deployments only).
+    pub instance_id: Option<String>,
 }
 
 impl AppState {
     pub fn new(config: ServerConfig) -> Self {
+        let instance_id = config.resolved_instance_id();
         Self {
             stores: RwLock::new(HashMap::new()),
             rollout_stores: RwLock::new(HashMap::new()),
             base_path: PathBuf::from(&config.data_dir),
+            instance_id,
         }
     }
 

crates/lance-context/src/unified_rollout.rs

@@ -1,5 +1,3 @@
-use std::collections::HashSet;
-
 use lance_context_api::{
     AddRolloutRequest, AddRolloutsResponse, ContextError, ContextResult, RolloutRecordDto,
     RolloutStoreApi,
@@ -29,17 +27,13 @@ impl RolloutStore {
     pub async fn open_with_options(
         uri: &str,
         storage_options: Option<std::collections::HashMap<String, String>>,
-        blob_columns: Option<Vec<String>>,
     ) -> Result<Self, ContextError> {
-        // `None` keeps the default blob offload of `binary_payload`; an explicit
-        // list (including empty) is taken verbatim.
-        let blob_columns: HashSet<String> = match blob_columns {
-            Some(cols) => cols.into_iter().collect(),
-            None => std::iter::once("binary_payload".to_string()).collect(),
-        };
         let options = RolloutStoreOptions {
             storage_options,
-            blob_columns,
+            // Embedded single-process use writes to the fallback shard. A
+            // multi-writer embedded deployment should thread a per-writer id
+            // through here (see `RolloutStoreOptions::shard_id`).
+            shard_id: None,
         };
         let store = LocalStore::open_with_options(uri, options)
             .await