commit a050efab334343bef8f5b618fcb2a8eea12b6da1 · ptr.pet/nsid-tracker

+37 -16

server/src/db/mod.rs

···

       402
        
               };

     

       403
        
       

     

       404
        
               // let mut ts = CLOCK.now();

     

       405
       -
               let mut current_item_count = 0;

     

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

     

       407
        
                   if current_item_count >= max_items {

     

       408
       -
                       return Ok(None);

     

       409
        
                   }

     

       0
        
       
     

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

     

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

     

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

     
···

       414
        
                       // tracing::info!(

     

       415
        
                       //     "stopped at block with timestamps {start_timestamp}..{end_timestamp} because {start_limit} is greater"

     

       416
        
                       // );

     

       417
       -
                       return Ok(None);

     

       418
        
                   }

     

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

     

       420
       -
                   current_item_count += decoder.item_count();

     

       421
        
                   // tracing::info!(

     

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

     

       423
        
                   //     ts.elapsed().as_nanos(),

     

       424
        
                   //     decoder.item_count()

     

       425
        
                   // );

     

       426
        
                   // ts = CLOCK.now();

     

       427
       -
                   Ok(Some(

     

       428
       -
                       decoder

     

       429
       -
                           .take_while(move |item| {

     

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

     

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

     

       0
        
       
     

       0
        
       
     

       432
        
                               })

     

       433
       -
                           })

     

       434
       -
                           .map(|res| res.map_err(AppError::from)),

     

       0
        
       
     

       435
        
                   ))

     

       436
        
               };

     

       437
        
       

     

       438
       -
               let blocks = handle

     

       439
        
                   .range(..end_key)

     

       0
        
       
     

       440
        
                   .rev()

     

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

     

       442
       -
                   .collect_vec();

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       443
        
       

     

       444
        
               tracing::info!(

     

       445
       -
                   "got blocks with size {}, item count {current_item_count}",

     

       446
        
                   blocks.len()

     

       447
        
               );

     

       448

···

       402
        
               };

     

       403
        
       

     

       404
        
               // let mut ts = CLOCK.now();

     

       405
       +
               let map_block = move |(res, current_item_count)| -> AppResult<(Option<_>, usize)> {

     

       0
        
       
     

       406
        
                   if current_item_count >= max_items {

     

       407
       +
                       return Ok((None, current_item_count));

     

       408
        
                   }

     

       409
       +
                   let (key, val) = res?;

     

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

     

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

     

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

     
···

       414
        
                       // tracing::info!(

     

       415
        
                       //     "stopped at block with timestamps {start_timestamp}..{end_timestamp} because {start_limit} is greater"

     

       416
        
                       // );

     

       417
       +
                       return Ok((None, current_item_count));

     

       418
        
                   }

     

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

     

       420
       +
                   let current_item_count = current_item_count + decoder.item_count();

     

       421
        
                   // tracing::info!(

     

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

     

       423
        
                   //     ts.elapsed().as_nanos(),

     

       424
        
                   //     decoder.item_count()

     

       425
        
                   // );

     

       426
        
                   // ts = CLOCK.now();

     

       427
       +
                   Ok((

     

       428
       +
                       Some(

     

       429
       +
                           decoder

     

       430
       +
                               .take_while(move |item| {

     

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

     

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

     

       433
       +
                                   })

     

       434
        
                               })

     

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

     

       436
       +
                       ),

     

       437
       +
                       current_item_count,

     

       438
        
                   ))

     

       439
        
               };

     

       440
        
       

     

       441
       +
               let (blocks, counted) = handle

     

       442
        
                   .range(..end_key)

     

       443
       +
                   .map(|res| res.map_err(AppError::from))

     

       444
        
                   .rev()

     

       445
       +
                   .fold_while(

     

       446
       +
                       (Vec::with_capacity(20), 0),

     

       447
       +
                       |(mut blocks, current_item_count), res| {

     

       448
       +
                           use itertools::FoldWhile::*;

     

       449
       +
       

     

       450
       +
                           match map_block((res, current_item_count)) {

     

       451
       +
                               Ok((Some(block), current_item_count)) => {

     

       452
       +
                                   blocks.push(Ok(block));

     

       453
       +
                                   Continue((blocks, current_item_count))

     

       454
       +
                               }

     

       455
       +
                               Ok((None, current_item_count)) => Done((blocks, current_item_count)),

     

       456
       +
                               Err(err) => {

     

       457
       +
                                   blocks.push(Err(err));

     

       458
       +
                                   Done((blocks, current_item_count))

     

       459
       +
                               }

     

       460
       +
                           }

     

       461
       +
                       },

     

       462
       +
                   )

     

       463
       +
                   .into_inner();

     

       464
        
       

     

       465
        
               tracing::info!(

     

       466
       +
                   "got blocks with size {}, item count {counted}",

     

       467
        
                   blocks.len()

     

       468
        
               );

     

       469