@@ -351,6 +351,7 @@ class SystemStore extends EventEmitter {
351
351
this . is_finished_initial_load = false ;
352
352
this . START_REFRESH_THRESHOLD = 10 * 60 * 1000 ;
353
353
this . FORCE_REFRESH_THRESHOLD = 60 * 60 * 1000 ;
354
+ this . READ_NEW_DATA_CONCURRENCY = config . READ_NEW_DATA_CONCURRENCY || 5 ;
354
355
this . _load_serial = new semaphore . Semaphore ( 1 , { warning_timeout : this . START_REFRESH_THRESHOLD } ) ;
355
356
for ( const col of COLLECTIONS ) {
356
357
db_client . instance ( ) . define_collection ( col ) ;
@@ -407,30 +408,33 @@ class SystemStore extends EventEmitter {
407
408
}
408
409
409
410
async load ( since ) {
411
+ const new_data = new SystemStoreData ( ) ;
412
+
410
413
// serializing load requests since we have to run a fresh load after the previous one will finish
411
414
// because it might not see the latest changes if we don't reload right after make_changes.
412
415
return this . _load_serial . surround ( async ( ) => {
413
416
try {
414
- dbg . log3 ( 'SystemStore: loading ...' ) ;
417
+ dbg . log3 ( 'SystemStore: loading ... this.last_update_time =' , this . last_update_time , ", since =" , since ) ;
415
418
416
419
// If we get a load request with an timestamp older then our last update time
417
420
// we ensure we load everyting from that timestamp by updating our last_update_time.
418
421
if ( ! _ . isUndefined ( since ) && since < this . last_update_time ) {
419
- dbg . log0 ( 'SystemStore.load: Got load request with a timestamp older then my last update time' ) ;
422
+ dbg . log0 ( 'SystemStore.load: Got load request with a timestamp' , since , ' older than my last update time', this . last_update_time ) ;
420
423
this . last_update_time = since ;
421
424
}
422
425
this . master_key_manager . load_root_key ( ) ;
423
- const new_data = new SystemStoreData ( ) ;
424
426
let millistamp = time_utils . millistamp ( ) ;
425
427
await this . _register_for_changes ( ) ;
426
428
await this . _read_new_data_from_db ( new_data ) ;
427
429
const secret = await os_utils . read_server_secret ( ) ;
428
430
this . _server_secret = secret ;
429
- dbg . log1 ( 'SystemStore: fetch took' , time_utils . millitook ( millistamp ) ) ;
430
- dbg . log1 ( 'SystemStore: fetch size' , size_utils . human_size ( JSON . stringify ( new_data ) . length ) ) ;
431
- dbg . log1 ( 'SystemStore: fetch data' , util . inspect ( new_data , {
432
- depth : 4
433
- } ) ) ;
431
+ if ( dbg . should_log ( 1 ) ) { //param should match below logs' level
432
+ dbg . log1 ( 'SystemStore: fetch took' , time_utils . millitook ( millistamp ) ) ;
433
+ dbg . log1 ( 'SystemStore: fetch size' , size_utils . human_size ( JSON . stringify ( new_data ) . length ) ) ;
434
+ dbg . log1 ( 'SystemStore: fetch data' , util . inspect ( new_data , {
435
+ depth : 4
436
+ } ) ) ;
437
+ }
434
438
this . old_db_data = this . _update_data_from_new ( this . old_db_data || { } , new_data ) ;
435
439
this . data = _ . cloneDeep ( this . old_db_data ) ;
436
440
millistamp = time_utils . millistamp ( ) ;
@@ -506,7 +510,7 @@ class SystemStore extends EventEmitter {
506
510
}
507
511
} ;
508
512
await db_client . instance ( ) . connect ( ) ;
509
- await P . map ( COLLECTIONS , async col => {
513
+ await P . map_with_concurrency ( this . READ_NEW_DATA_CONCURRENCY , COLLECTIONS , async col => {
510
514
const res = await db_client . instance ( ) . collection ( col . name )
511
515
. find ( newly_updated_query , {
512
516
projection : { last_update : 0 }
0 commit comments