feat(server): make db configurable, improve migration perf

server/src/db/mod.rs

···
     fmt::Debug,
     io::Cursor,
     ops::{Bound, Deref, RangeBounds},
-    path::Path,
+    path::{Path, PathBuf},
     time::Duration,
 };
 
···
     pub disk_size: u64,
 }
 
+pub struct DbConfig {
+    pub ks_config: fjall::Config,
+    pub min_block_size: usize,
+    pub max_block_size: usize,
+    pub max_last_activity: u64,
+}
+
+impl DbConfig {
+    pub fn path(mut self, path: impl AsRef<Path>) -> Self {
+        self.ks_config = fjall::Config::new(path);
+        self
+    }
+
+    pub fn ks(mut self, f: impl FnOnce(fjall::Config) -> fjall::Config) -> Self {
+        self.ks_config = f(self.ks_config);
+        self
+    }
+}
+
+impl Default for DbConfig {
+    fn default() -> Self {
+        Self {
+            ks_config: fjall::Config::default(),
+            min_block_size: 512,
+            max_block_size: 500_000,
+            max_last_activity: Duration::from_secs(10).as_nanos() as u64,
+        }
+    }
+}
+
 // counts is nsid -> NsidCounts
 // hits is tree per nsid: varint start time + varint end time -> block of hits
 pub struct Db {
-    ks: Keyspace,
+    pub cfg: DbConfig,
+    pub ks: Keyspace,
     counts: Partition,
     hits: scc::HashIndex<SmolStr, Arc<LexiconHandle>>,
     sync_pool: threadpool::ThreadPool,
     event_broadcaster: broadcast::Sender<(SmolStr, NsidCounts)>,
     eps: RateTracker<100>,
     cancel_token: CancellationToken,
-    pub min_block_size: usize,
-    pub max_block_size: usize,
-    pub max_last_activity: u64,
 }
 
 impl Db {
-    pub fn new(path: impl AsRef<Path>, cancel_token: CancellationToken) -> AppResult<Self> {
+    pub fn new(cfg: DbConfig, cancel_token: CancellationToken) -> AppResult<Self> {
         tracing::info!("opening db...");
-        let ks = Config::new(path).open()?;
+        let ks = cfg.ks_config.clone().open()?;
         Ok(Self {
+            cfg,
             hits: Default::default(),
             sync_pool: threadpool::Builder::new()
                 .num_threads(rayon::current_num_threads() * 2)
···
             event_broadcaster: broadcast::channel(1000).0,
             eps: RateTracker::new(Duration::from_secs(1)),
             cancel_token,
-            min_block_size: 512,
-            max_block_size: 500_000,
-            max_last_activity: Duration::from_secs(10).as_nanos() as u64,
         })
     }
 
···
         for (_, handle) in self.hits.iter(&_guard) {
             let mut nsid_data = Vec::with_capacity(2);
             let mut total_count = 0;
-            let is_too_old = handle.since_last_activity() > self.max_last_activity;
+            let is_too_old = handle.since_last_activity() > self.cfg.max_last_activity;
             // if we disconnect for a long time, we want to sync all of what we
             // have to avoid having many small blocks (even if we run compaction
             // later, it reduces work until we run compaction)
             let block_size = (is_too_old || all)
-                .then_some(self.max_block_size)
+                .then_some(self.cfg.max_block_size)
                 .unwrap_or_else(|| {
-                    self.max_block_size
-                        .min(self.min_block_size.max(handle.suggested_block_size()))
+                    self.cfg
+                        .max_block_size
+                        .min(self.cfg.min_block_size.max(handle.suggested_block_size()))
                 });
             let count = handle.item_count();
             let data_count = count / block_size;
···
     }
 
     pub fn major_compact(&self) -> AppResult<()> {
-        self.compact_all(self.max_block_size, .., true)?;
+        self.compact_all(self.cfg.max_block_size, .., true)?;
         let _guard = scc::ebr::Guard::new();
         for (_, handle) in self.hits.iter(&_guard) {
             handle.deref().major_compact()?;

server/src/main.rs

···
-use std::{ops::Deref, time::Duration};
+use std::{ops::Deref, time::Duration, u64};
 
 use itertools::Itertools;
 use rclite::Arc;
···
 
 use crate::{
     api::serve,
-    db::{Db, EventRecord},
+    db::{Db, DbConfig, EventRecord},
     error::AppError,
     jetstream::JetstreamClient,
     utils::{CLOCK, RelativeDateTime, get_time},
···
 
     let cancel_token = CancellationToken::new();
 
-    let db =
-        Arc::new(Db::new(".fjall_data", cancel_token.child_token()).expect("couldnt create db"));
+    let db = Arc::new(
+        Db::new(DbConfig::default(), cancel_token.child_token()).expect("couldnt create db"),
+    );
 
     rustls::crypto::ring::default_provider()
         .install_default()
···
                                 },
                                 "running compaction...",
                             );
-                            match db.compact_all(db.max_block_size, range, false) {
+                            match db.compact_all(db.cfg.max_block_size, range, false) {
                                 Ok(_) => (),
                                 Err(e) => tracing::error!("failed to compact db: {}", e),
                             }
···
 }
 
 fn debug() {
-    let db = Db::new(".fjall_data", CancellationToken::new()).expect("couldnt create db");
+    let db = Db::new(DbConfig::default(), CancellationToken::new()).expect("couldnt create db");
     let info = db.info().expect("cant get db info");
     println!("disk size: {}", info.disk_size);
     for (nsid, blocks) in info.nsids {
···
 }
 
 fn compact() {
-    let db = Db::new(".fjall_data", CancellationToken::new()).expect("couldnt create db");
+    let db = Db::new(DbConfig::default(), CancellationToken::new()).expect("couldnt create db");
     let info = db.info().expect("cant get db info");
     db.major_compact().expect("cant compact");
     std::thread::sleep(Duration::from_secs(5));
···
 fn migrate() {
     let cancel_token = CancellationToken::new();
     let from = Arc::new(
-        Db::new(".fjall_data_from", cancel_token.child_token()).expect("couldnt create db"),
+        Db::new(
+            DbConfig::default().path(".fjall_data_from"),
+            cancel_token.child_token(),
+        )
+        .expect("couldnt create db"),
     );
-    let to =
-        Arc::new(Db::new(".fjall_data_to", cancel_token.child_token()).expect("couldnt create db"));
+    let to = Arc::new(
+        Db::new(
+            DbConfig::default().path(".fjall_data_to").ks(|c| {
+                c.max_journaling_size(1024 * 1024 * 1024 * 8)
+                    .max_write_buffer_size(u64::MAX)
+                    .manual_journal_persist(true)
+            }),
+            cancel_token.child_token(),
+        )
+        .expect("couldnt create db"),
+    );
 
     let nsids = from.get_nsids().collect::<Vec<_>>();
     let eps_thread = std::thread::spawn({
···
     drop(from);
     tracing::info!("starting sync!!!");
     to.sync(true).expect("cant sync");
+    tracing::info!("persisting...");
+    to.ks
+        .persist(fjall::PersistMode::SyncAll)
+        .expect("cant persist");
     let total_time = start.elapsed();
     let write_per_second = total_count as f64 / (total_time - read_time).as_secs_f64();
     tracing::info!(