···

       
400
       
 
       
            return Either::Right(std::iter::empty());

     

       
401
       
 
       
        };

     

       
402
       
 
       


     

       
403
       
-
       
        let mut ts = CLOCK.now();

     

       
404
       
 
       
        let map_block = move |(key, val)| {

     

       
405
       
 
       
            let mut key_reader = Cursor::new(key);

     

       
406
       
 
       
            let start_timestamp = key_reader.read_varint::<u64>()?;

     

       
407
       
-
       
            let end_timestamp = key_reader.read_varint::<u64>()?;

     

       
408
       
 
       
            if start_timestamp < start_limit {

     

       
409
       
-
       
                tracing::info!(

     

       
410
       
-
       
                    "skipped block with timestamps {start_timestamp}..{end_timestamp} because {start_limit} is greater"

     

       
411
       
-
       
                );

     

       
412
       
 
       
                return Ok(None);

     

       
413
       
 
       
            } else {

     

       
414
       
-
       
                tracing::info!("using block with timestamp {start_timestamp}..{end_timestamp}");

     

       
415
       
 
       
            }

     

       
416
       
 
       
            let decoder = handle::ItemDecoder::new(Cursor::new(val), start_timestamp)?;

     

       
417
       
-
       
            tracing::info!(

     

       
418
       
-
       
                "took {}ns to get block with size {}",

     

       
419
       
-
       
                ts.elapsed().as_nanos(),

     

       
420
       
-
       
                decoder.item_count()

     

       
421
       
-
       
            );

     

       
422
       
-
       
            ts = CLOCK.now();

     

       
423
       
 
       
            Ok(Some(

     

       
424
       
 
       
                decoder

     

       
425
       
 
       
                    .take_while(move |item| {

     
···

       
431
       
 
       
            ))

     

       
432
       
 
       
        };

     

       
433
       
 
       


     

       
434
       
-
       
        Either::Left(

     

       
435
       
-
       
            handle

     

       
436
       
-
       
                .range(..end_key)

     

       
437
       
-
       
                .rev()

     

       
438
       
-
       
                .map_while(move |res| res.map_err(AppError::from).and_then(map_block).transpose())

     

       
439
       
-
       
                .collect::<Vec<_>>()

     

       
440
       
-
       
                .into_iter()

     

       
441
       
-
       
                .rev()

     

       
442
       
-
       
                .flatten()

     

       
443
       
-
       
                .flatten(),

     

       
444
       
-
       
        )

     

       
445
       
 
       
    }

     

       
446
       
 
       


     

       
447
       
 
       
    pub fn tracking_since(&self) -> AppResult<u64> {

     

···

       
400
       
 
       
            return Either::Right(std::iter::empty());

     

       
401
       
 
       
        };

     

       
402
       
 
       


     

       
403
       
+
       
        // let mut ts = CLOCK.now();

     

       
404
       
 
       
        let map_block = move |(key, val)| {

     

       
405
       
 
       
            let mut key_reader = Cursor::new(key);

     

       
406
       
 
       
            let start_timestamp = key_reader.read_varint::<u64>()?;

     

       
407
       
+
       
            // let end_timestamp = key_reader.read_varint::<u64>()?;

     

       
408
       
 
       
            if start_timestamp < start_limit {

     

       
409
       
+
       
                // tracing::info!(

     

       
410
       
+
       
                //     "skipped block with timestamps {start_timestamp}..{end_timestamp} because {start_limit} is greater"

     

       
411
       
+
       
                // );

     

       
412
       
 
       
                return Ok(None);

     

       
413
       
 
       
            } else {

     

       
414
       
+
       
                // tracing::info!("using block with timestamp {start_timestamp}..{end_timestamp}");

     

       
415
       
 
       
            }

     

       
416
       
 
       
            let decoder = handle::ItemDecoder::new(Cursor::new(val), start_timestamp)?;

     

       
417
       
+
       
            // tracing::info!(

     

       
418
       
+
       
            //     "took {}ns to get block with size {}",

     

       
419
       
+
       
            //     ts.elapsed().as_nanos(),

     

       
420
       
+
       
            //     decoder.item_count()

     

       
421
       
+
       
            // );

     

       
422
       
+
       
            // ts = CLOCK.now();

     

       
423
       
 
       
            Ok(Some(

     

       
424
       
 
       
                decoder

     

       
425
       
 
       
                    .take_while(move |item| {

     
···

       
431
       
 
       
            ))

     

       
432
       
 
       
        };

     

       
433
       
 
       


     

       
434
       
+
       
        let blocks = handle

     

       
435
       
+
       
            .range(..end_key)

     

       
436
       
+
       
            .rev()

     

       
437
       
+
       
            .map_while(move |res| res.map_err(AppError::from).and_then(map_block).transpose())

     

       
438
       
+
       
            .collect::<Vec<_>>();

     

       
439
       
+
       


     

       
440
       
+
       
        tracing::info!("returning {} blocks", blocks.len());

     

       
441
       
+
       


     

       
442
       
+
       
        Either::Left(blocks.into_iter().rev().flatten().flatten())

     

       
0
       
 
       

     

       
0
       
 
       

     

       
443
       
 
       
    }

     

       
444
       
 
       


     

       
445
       
 
       
    pub fn tracking_since(&self) -> AppResult<u64> {