···

       
402
       
402
       
 
       
        };

     

       
403
       
403
       
 
       
        tracing::info!("took {}ns to get handle", started_at.elapsed().as_nanos());

     

       
404
       
404
       
 
       


     

       
405
       
405
       
+
       
        let mut ts = CLOCK.now();

     

       
405
       
406
       
 
       
        let map_block = move |(key, val)| {

     

       
406
       
407
       
 
       
            let mut key_reader = Cursor::new(key);

     

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

     

       
408
       
409
       
 
       
            if start_timestamp < start_limit {

     

       
409
       
410
       
 
       
                return Ok(None);

     

       
410
       
411
       
 
       
            }

     

       
411
       
411
       
-
       
            let items = handle::ItemDecoder::new(Cursor::new(val), start_timestamp)?

     

       
412
       
412
       
-
       
                .take_while(move |item| {

     

       
413
       
413
       
-
       
                    item.as_ref().map_or(true, |item| {

     

       
414
       
414
       
-
       
                        item.timestamp <= end_limit && item.timestamp >= start_limit

     

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

     

       
413
       
413
       
+
       
            tracing::info!(

     

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

     

       
415
       
415
       
+
       
                ts.elapsed().as_nanos(),

     

       
416
       
416
       
+
       
                decoder.item_count()

     

       
417
       
417
       
+
       
            );

     

       
418
       
418
       
+
       
            ts = CLOCK.now();

     

       
419
       
419
       
+
       
            Ok(Some(

     

       
420
       
420
       
+
       
                decoder

     

       
421
       
421
       
+
       
                    .take_while(move |item| {

     

       
422
       
422
       
+
       
                        item.as_ref().map_or(true, |item| {

     

       
423
       
423
       
+
       
                            item.timestamp <= end_limit && item.timestamp >= start_limit

     

       
424
       
424
       
+
       
                        })

     

       
415
       
425
       
 
       
                    })

     

       
416
       
416
       
-
       
                })

     

       
417
       
417
       
-
       
                .map(|res| res.map_err(AppError::from));

     

       
418
       
418
       
-
       
            Ok(Some(items))

     

       
426
       
426
       
+
       
                    .map(|res| res.map_err(AppError::from)),

     

       
427
       
427
       
+
       
            ))

     

       
419
       
428
       
 
       
        };

     

       
420
       
429
       
 
       


     

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

     

       
422
       
430
       
 
       
        Either::Left(

     

       
423
       
431
       
 
       
            handle

     

       
424
       
432
       
 
       
                .range(..end_key)

     

       
425
       
433
       
 
       
                .rev()

     

       
426
       
426
       
-
       
                .inspect(|_| {

     

       
427
       
427
       
-
       
                    tracing::info!("took {}ns to get block", ts.elapsed().as_nanos());

     

       
428
       
428
       
-
       
                    ts = CLOCK.now();

     

       
429
       
429
       
-
       
                })

     

       
430
       
434
       
 
       
                .map_while(move |res| res.map_err(AppError::from).and_then(map_block).transpose())

     

       
431
       
435
       
 
       
                .collect::<Vec<_>>()

     

       
432
       
436
       
 
       
                .into_iter()