···

       
91
       
 
       
        self.eps.observe();

     

       
92
       
 
       
    }

     

       
93
       
 
       


     

       
94
       
-
       
    pub fn compact(&self, compact_to: usize, range: impl RangeBounds<u64>) -> AppResult<()> {

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
95
       
 
       
        let start_limit = match range.start_bound().cloned() {

     

       
96
       
 
       
            Bound::Included(start) => start,

     

       
97
       
 
       
            Bound::Excluded(start) => start.saturating_add(1),

     
···

       
117
       
 
       


     

       
118
       
 
       
        let start_blocks_size = blocks_to_compact.len();

     

       
119
       
 
       
        let keys_to_delete = blocks_to_compact.iter().map(|(key, _)| key);

     

       
120
       
-
       
        let all_items =

     

       
121
       
 
       
            blocks_to_compact

     

       
122
       
 
       
                .iter()

     

       
123
       
 
       
                .try_fold(Vec::new(), |mut acc, (key, value)| {

     
···

       
128
       
 
       
                    acc.append(&mut items);

     

       
129
       
 
       
                    AppResult::Ok(acc)

     

       
130
       
 
       
                })?;

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
131
       
 
       


     

       
132
       
 
       
        let new_blocks = all_items

     

       
133
       
 
       
            .into_iter()

     

···

       
91
       
 
       
        self.eps.observe();

     

       
92
       
 
       
    }

     

       
93
       
 
       


     

       
94
       
+
       
    pub fn compact(

     

       
95
       
+
       
        &self,

     

       
96
       
+
       
        compact_to: usize,

     

       
97
       
+
       
        range: impl RangeBounds<u64>,

     

       
98
       
+
       
        sort: bool,

     

       
99
       
+
       
    ) -> AppResult<()> {

     

       
100
       
 
       
        let start_limit = match range.start_bound().cloned() {

     

       
101
       
 
       
            Bound::Included(start) => start,

     

       
102
       
 
       
            Bound::Excluded(start) => start.saturating_add(1),

     
···

       
122
       
 
       


     

       
123
       
 
       
        let start_blocks_size = blocks_to_compact.len();

     

       
124
       
 
       
        let keys_to_delete = blocks_to_compact.iter().map(|(key, _)| key);

     

       
125
       
+
       
        let mut all_items =

     

       
126
       
 
       
            blocks_to_compact

     

       
127
       
 
       
                .iter()

     

       
128
       
 
       
                .try_fold(Vec::new(), |mut acc, (key, value)| {

     
···

       
133
       
 
       
                    acc.append(&mut items);

     

       
134
       
 
       
                    AppResult::Ok(acc)

     

       
135
       
 
       
                })?;

     

       
136
       
+
       


     

       
137
       
+
       
        if sort {

     

       
138
       
+
       
            all_items.sort_unstable_by_key(|e| e.timestamp);

     

       
139
       
+
       
        }

     

       
140
       
 
       


     

       
141
       
 
       
        let new_blocks = all_items

     

       
142
       
 
       
            .into_iter()

     

···

       
207
       
 
       
        nsid: impl AsRef<str>,

     

       
208
       
 
       
        max_count: usize,

     

       
209
       
 
       
        range: impl RangeBounds<u64>,

     

       
0
       
 
       

     

       
210
       
 
       
    ) -> AppResult<()> {

     

       
211
       
 
       
        let Some(handle) = self.get_handle(nsid) else {

     

       
212
       
 
       
            return Ok(());

     

       
213
       
 
       
        };

     

       
214
       
-
       
        handle.compact(max_count, range)

     

       
215
       
 
       
    }

     

       
216
       
 
       


     

       
217
       
 
       
    pub fn compact_all(

     

       
218
       
 
       
        &self,

     

       
219
       
 
       
        max_count: usize,

     

       
220
       
 
       
        range: impl RangeBounds<u64> + Clone,

     

       
0
       
 
       

     

       
221
       
 
       
    ) -> AppResult<()> {

     

       
222
       
 
       
        for nsid in self.get_nsids() {

     

       
223
       
-
       
            self.compact(nsid, max_count, range.clone())?;

     

       
224
       
 
       
        }

     

       
225
       
 
       
        Ok(())

     

       
226
       
 
       
    }

     

       
227
       
 
       


     

       
228
       
 
       
    pub fn major_compact(&self) -> AppResult<()> {

     

       
229
       
-
       
        self.compact_all(self.max_block_size, ..)?;

     

       
230
       
 
       
        let _guard = scc::ebr::Guard::new();

     

       
231
       
 
       
        for (_, handle) in self.hits.iter(&_guard) {

     

       
232
       
 
       
            handle.deref().major_compact()?;

     

···

       
207
       
 
       
        nsid: impl AsRef<str>,

     

       
208
       
 
       
        max_count: usize,

     

       
209
       
 
       
        range: impl RangeBounds<u64>,

     

       
210
       
+
       
        sort: bool,

     

       
211
       
 
       
    ) -> AppResult<()> {

     

       
212
       
 
       
        let Some(handle) = self.get_handle(nsid) else {

     

       
213
       
 
       
            return Ok(());

     

       
214
       
 
       
        };

     

       
215
       
+
       
        handle.compact(max_count, range, sort)

     

       
216
       
 
       
    }

     

       
217
       
 
       


     

       
218
       
 
       
    pub fn compact_all(

     

       
219
       
 
       
        &self,

     

       
220
       
 
       
        max_count: usize,

     

       
221
       
 
       
        range: impl RangeBounds<u64> + Clone,

     

       
222
       
+
       
        sort: bool,

     

       
223
       
 
       
    ) -> AppResult<()> {

     

       
224
       
 
       
        for nsid in self.get_nsids() {

     

       
225
       
+
       
            self.compact(nsid, max_count, range.clone(), sort)?;

     

       
226
       
 
       
        }

     

       
227
       
 
       
        Ok(())

     

       
228
       
 
       
    }

     

       
229
       
 
       


     

       
230
       
 
       
    pub fn major_compact(&self) -> AppResult<()> {

     

       
231
       
+
       
        self.compact_all(self.max_block_size, .., true)?;

     

       
232
       
 
       
        let _guard = scc::ebr::Guard::new();

     

       
233
       
 
       
        for (_, handle) in self.hits.iter(&_guard) {

     

       
234
       
 
       
            handle.deref().major_compact()?;

     

···

       
155
       
 
       
                                },

     

       
156
       
 
       
                                "running compaction...",

     

       
157
       
 
       
                            );

     

       
158
       
-
       
                            match db.compact_all(db.max_block_size, range) {

     

       
159
       
 
       
                                Ok(_) => (),

     

       
160
       
 
       
                                Err(e) => tracing::error!("failed to compact db: {}", e),

     

       
161
       
 
       
                            }

     

···

       
155
       
 
       
                                },

     

       
156
       
 
       
                                "running compaction...",

     

       
157
       
 
       
                            );

     

       
158
       
+
       
                            match db.compact_all(db.max_block_size, range, false) {

     

       
159
       
 
       
                                Ok(_) => (),

     

       
160
       
 
       
                                Err(e) => tracing::error!("failed to compact db: {}", e),

     

       
161
       
 
       
                            }

     

···

       
290
       
 
       
        }

     

       
291
       
 
       
    }

     

       
292
       
 
       


     

       
293
       
-
       
    pub fn ago(duration: Duration) -> Self {

     

       
294
       
-
       
        Self::new(duration, TimeDirection::Backwards)

     

       
295
       
-
       
    }

     

       
296
       
-
       


     

       
297
       
 
       
    pub fn from_now(duration: Duration) -> Self {

     

       
298
       
 
       
        let cur = get_time();

     

       
299
       
 
       
        if duration > cur {

     

···

       
290
       
 
       
        }

     

       
291
       
 
       
    }

     

       
292
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
293
       
 
       
    pub fn from_now(duration: Duration) -> Self {

     

       
294
       
 
       
        let cur = get_time();

     

       
295
       
 
       
        if duration > cur {