···

       
76
       
 
       
        self.buf.lock().len()

     

       
77
       
 
       
    }

     

       
78
       
 
       


     

       
79
       
-
       
    pub fn since_last_activity(&self) -> u64 {

     

       
80
       
-
       
        CLOCK.delta_as_nanos(self.last_insert.load(AtomicOrdering::Relaxed), CLOCK.raw())

     

       
0
       
 
       

     

       
0
       
 
       

     

       
81
       
 
       
    }

     

       
82
       
 
       


     

       
83
       
 
       
    pub fn suggested_block_size(&self) -> usize {

     

···

       
76
       
 
       
        self.buf.lock().len()

     

       
77
       
 
       
    }

     

       
78
       
 
       


     

       
79
       
+
       
    pub fn since_last_activity(&self) -> Duration {

     

       
80
       
+
       
        Duration::from_nanos(

     

       
81
       
+
       
            CLOCK.delta_as_nanos(self.last_insert.load(AtomicOrdering::Relaxed), CLOCK.raw()),

     

       
82
       
+
       
        )

     

       
83
       
 
       
    }

     

       
84
       
 
       


     

       
85
       
 
       
    pub fn suggested_block_size(&self) -> usize {

     

···

       
3
       
 
       
    fmt::Debug,

     

       
4
       
 
       
    io::Cursor,

     

       
5
       
 
       
    ops::{Bound, Deref, RangeBounds},

     

       
6
       
-
       
    path::{Path, PathBuf},

     

       
7
       
 
       
    time::Duration,

     

       
8
       
 
       
};

     

       
9
       
 
       


     

       
10
       
 
       
use byteview::StrView;

     

       
11
       
-
       
use fjall::{Config, Keyspace, Partition, PartitionCreateOptions};

     

       
12
       
 
       
use itertools::{Either, Itertools};

     

       
13
       
-
       
use rayon::iter::{IndexedParallelIterator, IntoParallelIterator, ParallelIterator};

     

       
14
       
 
       
use rclite::Arc;

     

       
15
       
 
       
use rkyv::{Archive, Deserialize, Serialize, rancor::Error};

     

       
16
       
 
       
use smol_str::{SmolStr, ToSmolStr};

     
···

       
79
       
 
       
    pub ks_config: fjall::Config,

     

       
80
       
 
       
    pub min_block_size: usize,

     

       
81
       
 
       
    pub max_block_size: usize,

     

       
82
       
-
       
    pub max_last_activity: u64,

     

       
83
       
 
       
}

     

       
84
       
 
       


     

       
85
       
 
       
impl DbConfig {

     
···

       
98
       
 
       
    fn default() -> Self {

     

       
99
       
 
       
        Self {

     

       
100
       
 
       
            ks_config: fjall::Config::default(),

     

       
101
       
-
       
            min_block_size: 512,

     

       
102
       
-
       
            max_block_size: 500_000,

     

       
103
       
-
       
            max_last_activity: Duration::from_secs(10).as_nanos() as u64,

     

       
104
       
 
       
        }

     

       
105
       
 
       
    }

     

       
106
       
 
       
}

     
···

       
114
       
 
       
    hits: scc::HashIndex<SmolStr, Arc<LexiconHandle>>,

     

       
115
       
 
       
    sync_pool: threadpool::ThreadPool,

     

       
116
       
 
       
    event_broadcaster: broadcast::Sender<(SmolStr, NsidCounts)>,

     

       
117
       
-
       
    eps: RateTracker<100>,

     

       
118
       
 
       
    cancel_token: CancellationToken,

     

       
119
       
 
       
}

     

       
120
       
 
       


     
···

       
229
       
 
       
                    self.sync_pool

     

       
230
       
 
       
                        .execute(move || match handle.insert(block.key, block.data) {

     

       
231
       
 
       
                            Ok(_) => {

     

       
232
       
-
       
                                tracing::info!("{}: [{i}] synced {}", block.written, handle.nsid())

     

       
233
       
 
       
                            }

     

       
234
       
 
       
                            Err(err) => tracing::error!("failed to sync block: {}", err),

     

       
235
       
 
       
                        });

     

···

       
3
       
 
       
    fmt::Debug,

     

       
4
       
 
       
    io::Cursor,

     

       
5
       
 
       
    ops::{Bound, Deref, RangeBounds},

     

       
6
       
+
       
    path::Path,

     

       
7
       
 
       
    time::Duration,

     

       
8
       
 
       
};

     

       
9
       
 
       


     

       
10
       
 
       
use byteview::StrView;

     

       
11
       
+
       
use fjall::{Keyspace, Partition, PartitionCreateOptions};

     

       
12
       
 
       
use itertools::{Either, Itertools};

     

       
13
       
+
       
use rayon::iter::{IntoParallelIterator, ParallelIterator};

     

       
14
       
 
       
use rclite::Arc;

     

       
15
       
 
       
use rkyv::{Archive, Deserialize, Serialize, rancor::Error};

     

       
16
       
 
       
use smol_str::{SmolStr, ToSmolStr};

     
···

       
79
       
 
       
    pub ks_config: fjall::Config,

     

       
80
       
 
       
    pub min_block_size: usize,

     

       
81
       
 
       
    pub max_block_size: usize,

     

       
82
       
+
       
    pub max_last_activity: Duration,

     

       
83
       
 
       
}

     

       
84
       
 
       


     

       
85
       
 
       
impl DbConfig {

     
···

       
98
       
 
       
    fn default() -> Self {

     

       
99
       
 
       
        Self {

     

       
100
       
 
       
            ks_config: fjall::Config::default(),

     

       
101
       
+
       
            min_block_size: 1000,

     

       
102
       
+
       
            max_block_size: 1_000_000,

     

       
103
       
+
       
            max_last_activity: Duration::from_secs(10),

     

       
104
       
 
       
        }

     

       
105
       
 
       
    }

     

       
106
       
 
       
}

     
···

       
114
       
 
       
    hits: scc::HashIndex<SmolStr, Arc<LexiconHandle>>,

     

       
115
       
 
       
    sync_pool: threadpool::ThreadPool,

     

       
116
       
 
       
    event_broadcaster: broadcast::Sender<(SmolStr, NsidCounts)>,

     

       
117
       
+
       
    eps: RateTracker<100>, // 100 millis buckets

     

       
118
       
 
       
    cancel_token: CancellationToken,

     

       
119
       
 
       
}

     

       
120
       
 
       


     
···

       
229
       
 
       
                    self.sync_pool

     

       
230
       
 
       
                        .execute(move || match handle.insert(block.key, block.data) {

     

       
231
       
 
       
                            Ok(_) => {

     

       
232
       
+
       
                                tracing::info!("{}: [{i}] synced {}", handle.nsid(), block.written)

     

       
233
       
 
       
                            }

     

       
234
       
 
       
                            Err(err) => tracing::error!("failed to sync block: {}", err),

     

       
235
       
 
       
                        });

     

···

       
280
       
 
       
    );

     

       
281
       
 
       


     

       
282
       
 
       
    let nsids = from.get_nsids().collect::<Vec<_>>();

     

       
283
       
-
       
    let eps_thread = std::thread::spawn({

     

       
284
       
 
       
        let to = to.clone();

     

       
285
       
 
       
        move || {

     

       
286
       
 
       
            loop {

     

···

       
280
       
 
       
    );

     

       
281
       
 
       


     

       
282
       
 
       
    let nsids = from.get_nsids().collect::<Vec<_>>();

     

       
283
       
+
       
    let _eps_thread = std::thread::spawn({

     

       
284
       
 
       
        let to = to.clone();

     

       
285
       
 
       
        move || {

     

       
286
       
 
       
            loop {