diff --git a/lib/crates/fabro-api/src/serve.rs b/lib/crates/fabro-api/src/serve.rs index 491b256f5..1e85c5111 100644 --- a/lib/crates/fabro-api/src/serve.rs +++ b/lib/crates/fabro-api/src/serve.rs @@ -152,7 +152,11 @@ pub async fn serve_command(args: ServeArgs, styles: &'static Styles) -> anyhow:: let store_path = data_dir.join("store"); std::fs::create_dir_all(&store_path)?; let object_store = Arc::new(LocalFileSystem::new_with_prefix(&store_path)?); - let store = Arc::new(fabro_store::SlateStore::new(object_store, "")); + let store = Arc::new(fabro_store::SlateStore::new( + object_store, + "", + Duration::from_millis(5), + )); let state = create_app_state_with_store( db, factory, diff --git a/lib/crates/fabro-cli/src/store.rs b/lib/crates/fabro-cli/src/store.rs index f1de4a0fa..6b3e31ff8 100644 --- a/lib/crates/fabro-cli/src/store.rs +++ b/lib/crates/fabro-cli/src/store.rs @@ -1,5 +1,6 @@ use std::path::Path; use std::sync::Arc; +use std::time::Duration; use anyhow::Result; use fabro_store::{RunStore, SlateStore, Store}; @@ -9,7 +10,11 @@ pub(crate) fn build_store(storage_dir: &Path) -> Result> { let store_path = storage_dir.join("store"); std::fs::create_dir_all(&store_path)?; let object_store = Arc::new(LocalFileSystem::new_with_prefix(&store_path)?); - Ok(Arc::new(SlateStore::new(object_store, ""))) + Ok(Arc::new(SlateStore::new( + object_store, + "", + Duration::from_millis(5), + ))) } pub(crate) async fn open_run_reader( diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index 55712eada..19a4d6ba4 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -3,6 +3,7 @@ mod run_store; use std::collections::HashMap; use std::sync::Arc; +use std::time::Duration; use async_trait::async_trait; use chrono::{DateTime, Utc}; @@ -10,7 +11,7 @@ use futures::TryStreamExt; use object_store::ObjectStore; use object_store::path::Path; use slatedb::DbReader; -use slatedb::config::DbReaderOptions; +use slatedb::config::{DbReaderOptions, Settings}; use tokio::sync::Mutex; use crate::keys; @@ -21,6 +22,7 @@ use run_store::{SlateRunStore, SlateRunStoreInner}; pub struct SlateStore { object_store: Arc, base_prefix: String, + flush_interval: Duration, active_runs: Arc>>>, } @@ -28,15 +30,21 @@ impl std::fmt::Debug for SlateStore { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("SlateStore") .field("base_prefix", &self.base_prefix) + .field("flush_interval", &self.flush_interval) .finish_non_exhaustive() } } impl SlateStore { - pub fn new(object_store: Arc, base_prefix: impl Into) -> Self { + pub fn new( + object_store: Arc, + base_prefix: impl Into, + flush_interval: Duration, + ) -> Self { Self { object_store, base_prefix: normalize_base_prefix(base_prefix.into()), + flush_interval, active_runs: Arc::new(Mutex::new(HashMap::new())), } } @@ -46,7 +54,15 @@ impl SlateStore { } async fn open_db(&self, db_prefix: &str) -> Result { - Ok(slatedb::Db::open(db_prefix.to_string(), self.object_store.clone()).await?) + Ok( + slatedb::Db::builder(db_prefix.to_string(), self.object_store.clone()) + .with_settings(Settings { + flush_interval: Some(self.flush_interval), + ..Settings::default() + }) + .build() + .await?, + ) } async fn open_reader(&self, db_prefix: &str) -> Result { @@ -382,7 +398,7 @@ mod tests { fn make_store() -> (Arc, SlateStore) { let object_store: Arc = Arc::new(InMemory::new()); - let store = SlateStore::new(object_store.clone(), "runs/"); + let store = SlateStore::new(object_store.clone(), "runs/", Duration::from_millis(5)); (object_store, store) } @@ -498,7 +514,12 @@ mod tests { record: &CatalogRecord, include_init: bool, ) -> slatedb::Db { - let db = slatedb::Db::open(record.db_prefix.clone(), object_store) + let db = slatedb::Db::builder(record.db_prefix.clone(), object_store) + .with_settings(slatedb::config::Settings { + flush_interval: Some(Duration::from_millis(5)), + ..slatedb::config::Settings::default() + }) + .build() .await .unwrap(); if include_init { @@ -858,7 +879,12 @@ mod tests { let (object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); let db_prefix = catalog::db_prefix("runs/", created_at, "run-1"); - let db = slatedb::Db::open(db_prefix.clone(), object_store) + let db = slatedb::Db::builder(db_prefix.clone(), object_store) + .with_settings(slatedb::config::Settings { + flush_interval: Some(Duration::from_millis(5)), + ..slatedb::config::Settings::default() + }) + .build() .await .unwrap(); let mismatched = CatalogRecord {