fix(server): persist seal_on_add and sweep every store kind

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

@@ -38,6 +38,16 @@ use lance_context_api::schema_spec::{SchemaSpec, ID_COLUMN};
 /// reopened without the caller re-declaring its columns.
 const SCHEMA_SPEC_KEY: &str = "lance-context:schema_spec";
 
+/// Schema-metadata key holding the store's seal-on-add mode.
+///
+/// Persisted for the same reason as the schema: it is a property *of the
+/// store*, chosen at creation, not of whoever happens to open it. Without this
+/// a store created with `seal_on_add: true` silently loses read-your-write the
+/// first time it is reopened (after an LRU eviction, or a process restart) —
+/// no error, no data loss, just "sometimes I can't read what I just wrote",
+/// which is close to undiagnosable in production.
+const SEAL_ON_ADD_KEY: &str = "lance-context:seal_on_add";
+
 /// Configuration for opening a [`GenericStore`].
 #[derive(Debug, Clone, Default)]
 pub struct GenericStoreOptions {
@@ -71,6 +81,9 @@ pub struct GenericStore {
     /// not against `spec.to_arrow()`, so a dataset whose physical field order
     /// differs still writes correctly.
     schema: Arc<Schema>,
+    /// Whether `add` seals before returning. Resolved from the dataset on a
+    /// reopen, from the caller on a create.
+    seal_on_add: bool,
 }
 
 impl GenericStore {
@@ -116,7 +129,7 @@ impl GenericStore {
         // key column against whatever it is given, and an empty placeholder
         // would fail that check.
         let create_schema = match &spec {
-            Some(spec) => Arc::new(schema_with_spec(spec)?),
+            Some(spec) => Arc::new(schema_with_spec(spec, options.seal_on_add)?),
             None if create_if_missing => {
                 return Err(ArrowError::SchemaError(
                     "a schema is required to create a store".to_string(),
@@ -134,6 +147,14 @@ impl GenericStore {
             }
         };
 
+        // A reopen honors the mode the store was *created* with; only a create
+        // takes it from the caller. Read it before opening, since the base
+        // needs it up front.
+        let seal_on_add = match &spec {
+            Some(_) => options.seal_on_add,
+            None => seal_on_add_from_schema(&create_schema).unwrap_or(options.seal_on_add),
+        };
+
         let base = StorageBase::open(
             uri,
             StorageBaseOptions {
@@ -148,7 +169,7 @@ impl GenericStore {
                 // User schemas are immutable in v1, so there is nothing to
                 // evolve an older base table to.
                 latest_schema: None,
-                seal_on_put: options.seal_on_add,
+                seal_on_put: seal_on_add,
             },
             create_if_missing,
         )
@@ -172,6 +193,7 @@ impl GenericStore {
             base,
             spec: persisted,
             schema,
+            seal_on_add,
         })
     }
 
@@ -181,6 +203,13 @@ impl GenericStore {
         &self.spec
     }
 
+    /// Whether `add` seals before returning, so its rows are immediately
+    /// readable. Persisted with the store, so a reopen preserves it.
+    #[must_use]
+    pub fn seal_on_add(&self) -> bool {
+        self.seal_on_add
+    }
+
     /// URI of the underlying Lance dataset.
     #[must_use]
     pub fn uri(&self) -> &str {
@@ -358,17 +387,28 @@ impl GenericStore {
     }
 }
 
-/// Attach the serialized spec to the Arrow schema so it round-trips on open.
-fn schema_with_spec(spec: &SchemaSpec) -> Result<Schema, ArrowError> {
+/// Attach the serialized spec and the seal mode to the Arrow schema, so both
+/// round-trip on open.
+fn schema_with_spec(spec: &SchemaSpec, seal_on_add: bool) -> Result<Schema, ArrowError> {
     let schema = spec.to_arrow()?;
     let encoded = serde_json::to_string(spec)
         .map_err(|error| ArrowError::SchemaError(format!("could not serialize schema: {error}")))?;
 
     let mut metadata = schema.metadata().clone();
     metadata.insert(SCHEMA_SPEC_KEY.to_string(), encoded);
+    metadata.insert(SEAL_ON_ADD_KEY.to_string(), seal_on_add.to_string());
     Ok(schema.with_metadata(metadata))
 }
 
+/// Read the persisted seal mode. `None` for a store written before this was
+/// persisted, which then falls back to the caller's option.
+fn seal_on_add_from_schema(schema: &Schema) -> Option<bool> {
+    schema
+        .metadata()
+        .get(SEAL_ON_ADD_KEY)
+        .and_then(|raw| raw.parse().ok())
+}
+
 /// Read the spec back out of a dataset's schema metadata.
 fn spec_from_schema(schema: &Schema) -> Result<SchemaSpec, ArrowError> {
     let encoded = schema.metadata().get(SCHEMA_SPEC_KEY).ok_or_else(|| {
@@ -494,6 +534,63 @@ mod tests {
         });
     }
 
+    #[test]
+    fn seal_mode_survives_a_reopen() {
+        // The mode is a property of the store, not of whoever opens it. Before
+        // it was persisted, a reopen silently reverted to the caller's default
+        // and the store lost read-your-write with no error.
+        let dir = TempDir::new().unwrap();
+        let uri = dir.path().to_string_lossy().to_string();
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        rt.block_on(async {
+            {
+                let store = GenericStore::open(&uri, spec(), sealing()).await.unwrap();
+                assert!(store.seal_on_add());
+            }
+
+            // Reopen with the *opposite* default: the persisted value wins.
+            let reopened = GenericStore::open_existing(&uri, GenericStoreOptions::default())
+                .await
+                .unwrap();
+            assert!(
+                reopened.seal_on_add(),
+                "a store created with seal_on_add must keep it across a reopen"
+            );
+
+            // And it is behavioral, not just a flag: the write is visible
+            // without an explicit flush.
+            reopened.add(&[row(json!({"id": "r1"}))]).await.unwrap();
+            assert_eq!(reopened.list(None, None).await.unwrap().len(), 1);
+        });
+    }
+
+    #[test]
+    fn deferred_seal_mode_also_survives_a_reopen() {
+        let dir = TempDir::new().unwrap();
+        let uri = dir.path().to_string_lossy().to_string();
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        rt.block_on(async {
+            {
+                GenericStore::open(&uri, spec(), GenericStoreOptions::default())
+                    .await
+                    .unwrap();
+            }
+
+            // Reopen with the opposite default; the persisted `false` wins.
+            let reopened = GenericStore::open_existing(&uri, sealing()).await.unwrap();
+            assert!(!reopened.seal_on_add());
+
+            reopened.add(&[row(json!({"id": "r1"}))]).await.unwrap();
+            assert_eq!(
+                reopened.list(None, None).await.unwrap().len(),
+                0,
+                "a deferred-seal store must not start sealing after a reopen"
+            );
+            reopened.flush().await.unwrap();
+            assert_eq!(reopened.list(None, None).await.unwrap().len(), 1);
+        });
+    }
+
     #[test]
     fn reopening_with_a_conflicting_schema_is_rejected() {
         // Silently reinterpreting existing data under a new schema is the

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

@@ -44,16 +44,29 @@ pub struct ServerConfig {
     /// regardless of count, so stale generations are folded in even on
     /// low-traffic shards that never cross the count threshold. `0` (the default)
     /// disables the timer.
+    ///
+    /// # Applies to every store kind, despite the name
+    ///
+    /// The sweeper visits rollout, datagen and generic stores alike — every
+    /// MemWAL-backed store accumulates generations, and only a merge reclaims
+    /// them. The `ROLLOUT_` prefix is historical: these variables predate the
+    /// other store kinds, and renaming them would silently break existing
+    /// deployment manifests for no functional gain.
     #[arg(long, env = "ROLLOUT_CLEANUP_INTERVAL_SECS", default_value = "0")]
     pub rollout_cleanup_interval_secs: u64,
 
-    /// Interval, in seconds, at which the sweeper flushes each resident rollout
-    /// store's active MemWAL memtable into a queryable generation. Rollout
-    /// appends are durable on return (the WAL entry is persisted to object
-    /// storage) but are not visible to reads until the memtable is flushed, so
+    /// Interval, in seconds, at which the sweeper flushes each resident store's
+    /// active MemWAL memtable into a queryable generation. A deferred-seal
+    /// append is durable on return (the WAL entry is persisted to object
+    /// storage) but is not visible to reads until the memtable is flushed, so
     /// this interval bounds read-after-write latency. Decoupling flush from the
-    /// append path is what lets concurrent appends run without serializing behind
-    /// a per-append seal. Default `30`.
+    /// append path is what lets concurrent appends run without serializing
+    /// behind a per-append seal. Default `30`.
+    ///
+    /// Like the cleanup interval, this applies to **every** store kind despite
+    /// the `ROLLOUT_` prefix. Rollout and generic stores default to a deferred
+    /// seal and depend on it; datagen seals on each append, so its pass is a
+    /// no-op in steady state.
     ///
     /// `0` disables periodic flush, leaving the cleanup sweeper
     /// (`ROLLOUT_CLEANUP_INTERVAL_SECS`) as the only thing that seals memtables
@@ -120,3 +133,57 @@ impl ServerConfig {
             .filter(|value| !value.is_empty())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use clap::CommandFactory;
+
+    /// Every tunable must stay an *optional flag* with a default.
+    ///
+    /// Regression guard: dropping a field's `#[arg(..)]` attribute silently
+    /// turns it into a required positional argument, so the binary still
+    /// compiles and every unit test still passes — it just refuses to start
+    /// with the documented flags. That is only caught by actually running the
+    /// server, which nothing in the Rust suite does.
+    #[test]
+    fn every_config_field_is_an_optional_flag() {
+        let command = ServerConfig::command();
+
+        let positionals: Vec<_> = command
+            .get_positionals()
+            .map(|arg| arg.get_id().to_string())
+            .collect();
+        assert!(
+            positionals.is_empty(),
+            "config fields must be flags, not positional args; found {positionals:?} \
+             (a missing #[arg(..)] attribute does this)"
+        );
+
+        for arg in command.get_arguments() {
+            if arg.get_id() == "help" || arg.get_id() == "version" {
+                continue;
+            }
+            assert!(
+                arg.get_long().is_some(),
+                "'{}' has no long flag",
+                arg.get_id()
+            );
+            assert!(
+                arg.get_default_values().len() == 1 || !arg.is_required_set(),
+                "'{}' must have a default or be optional",
+                arg.get_id()
+            );
+        }
+    }
+
+    /// The server must start with no arguments beyond a data directory — the
+    /// documented defaults have to actually be defaults.
+    #[test]
+    fn parses_with_only_a_data_dir() {
+        let config = ServerConfig::try_parse_from(["lance-context-server", "--data-dir", "/tmp/x"])
+            .expect("the documented minimal invocation must parse");
+        assert_eq!(config.rollout_flush_interval_secs, 30);
+        assert_eq!(config.rollout_cleanup_interval_secs, 0);
+    }
+}

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

@@ -2,6 +2,7 @@ mod config;
 mod error;
 mod routes;
 mod state;
+mod sweeper;
 
 use std::sync::Arc;
 
@@ -58,9 +59,10 @@ async fn main() {
     if state.rollout_flush_interval_secs == 0 && state.rollout_cleanup_interval_secs == 0 {
         tracing::warn!(
             "ROLLOUT_FLUSH_INTERVAL_SECS=0 and ROLLOUT_CLEANUP_INTERVAL_SECS=0: \
-             nothing will seal MemWAL memtables, so rollout appends will be durable \
-             but invisible to reads until this process restarts. Set at least one \
-             of them to a non-zero interval."
+             nothing will seal MemWAL memtables, so deferred-seal appends (rollout \
+             and generic stores) will be durable but invisible to reads until this \
+             process restarts. Pending WAL generations will also never be merged, \
+             for every store kind. Set at least one of them to a non-zero interval."
         );
     }
 

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

@@ -464,6 +464,105 @@ mod tests {
         assert!(matches!(err, AppError::NotFound(_)), "{err:?}");
     }
 
+    /// The bug this guards: `seal_on_add` used to live only in the open
+    /// options, so a store created with it lost read-your-write the first time
+    /// it was evicted from the LRU and reopened — silently, with no error and
+    /// no data loss, just "sometimes I can't read what I just wrote".
+    #[tokio::test]
+    async fn seal_mode_survives_eviction_and_reopen() {
+        let (state, _dir) = test_state().await;
+        let _ = create_generic_store(
+            State(state.clone()),
+            Json(CreateGenericStoreRequest {
+                name: "s1".to_string(),
+                schema: spec(),
+                storage_options: None,
+                seal_on_add: true,
+            }),
+        )
+        .await
+        .unwrap();
+
+        // Evict, forcing the next request to reopen from object storage. This
+        // is exactly what the LRU does under capacity pressure.
+        state.generic_stores.lock().await.pop("s1");
+
+        let _ = add_rows(
+            State(state.clone()),
+            Path("s1".to_string()),
+            Json(AddRowsRequest {
+                rows: vec![row(serde_json::json!({"id": "r1"}))],
+            }),
+        )
+        .await
+        .unwrap();
+
+        let Json(listed) = list_rows(
+            State(state),
+            Path("s1".to_string()),
+            Query(ListRowsQuery {
+                limit: None,
+                offset: None,
+                filter: None,
+            }),
+        )
+        .await
+        .unwrap();
+        assert_eq!(
+            listed.rows.len(),
+            1,
+            "a reopened store must keep the seal mode it was created with"
+        );
+    }
+
+    /// The mirror case: a store created with the deferred default must not
+    /// start sealing after a reopen either.
+    #[tokio::test]
+    async fn deferred_seal_mode_also_survives_eviction() {
+        let (state, _dir) = test_state().await;
+        let _ = create_generic_store(
+            State(state.clone()),
+            Json(CreateGenericStoreRequest {
+                name: "s1".to_string(),
+                schema: spec(),
+                storage_options: None,
+                seal_on_add: false,
+            }),
+        )
+        .await
+        .unwrap();
+
+        state.generic_stores.lock().await.pop("s1");
+
+        let _ = add_rows(
+            State(state.clone()),
+            Path("s1".to_string()),
+            Json(AddRowsRequest {
+                rows: vec![row(serde_json::json!({"id": "r1"}))],
+            }),
+        )
+        .await
+        .unwrap();
+
+        let query = || ListRowsQuery {
+            limit: None,
+            offset: None,
+            filter: None,
+        };
+        let Json(listed) = list_rows(State(state.clone()), Path("s1".to_string()), Query(query()))
+            .await
+            .unwrap();
+        assert_eq!(listed.rows.len(), 0, "deferred seal must stay deferred");
+
+        flush_generic_store(State(state.clone()), Path("s1".to_string()))
+            .await
+            .unwrap();
+        let Json(after) = list_rows(State(state), Path("s1".to_string()), Query(query()))
+            .await
+            .unwrap();
+        assert_eq!(after.rows.len(), 1);
+    }
+
     #[tokio::test]
     async fn deferred_seal_needs_a_flush_and_wal_merges() {
         let (state, _dir) = test_state().await;