Builder chunk batches locking (mainly to fix nodes_monitor working on same chunks)

src/server/bg_services/scrubber.js

@@ -5,6 +5,9 @@ const dbg = require('../../util/debug_module')(__filename);
 const config = require('../../../config');
 const md_store = require('../object_services/md_store');
 const map_builder = require('../object_services/map_builder');
+const system_store = require('../system_services/system_store').get_instance();
+const system_server_utils = require('../utils/system_server_utils');
+
 
 /**
  *
@@ -16,33 +19,42 @@ const map_builder = require('../object_services/map_builder');
  *
  */
 function background_worker() {
+    if (!system_store.is_finished_initial_load) {
+        dbg.log0('System did not finish initial load');
+        return;
+    }
+
+    const system = system_store.data.systems[0];
+    if (!system || system_server_utils.system_in_maintenance(system._id)) return;
+
     var self = this;
     return P.fcall(function() {
             var now = Date.now();
             var query = {
                 $and: [{
-                    $or: [{
-                        last_build: null
-                    }, {
-                        last_build: {
-                            $lt: new Date(now - config.REBUILD_LAST_BUILD_BACKOFF)
-                        }
-                    }]
-                }, {
-                    $or: [{
-                        building: null
+                        $or: [{
+                            last_build: null
+                        }, {
+                            last_build: {
+                                $lt: new Date(now - config.REBUILD_LAST_BUILD_BACKOFF)
+                            }
+                        }]
                     }, {
-                        building: {
-                            $lt: new Date(now - config.REBUILD_BUILDING_MODE_BACKOFF)
+                        $or: [{
+                            building: null
+                        }, {
+                            building: {
+                                $lt: new Date(now - config.REBUILD_BUILDING_MODE_BACKOFF)
+                            }
+                        }]
+                    },
+                    //ignore old chunks without buckets
+                    {
+                        bucket: {
+                            $exists: true
                         }
-                    }]
-                },
-                //ignore old chunks without buckets
-                {
-                    bucket: {
-                        $exists: true
                     }
-                }]
+                ]
             };
             if (self.last_chunk_id) {
                 query._id = {
@@ -54,29 +66,24 @@ function background_worker() {
             query.deleted = null;
 
             return P.resolve(md_store.DataChunk.find(query)
+                .select('_id')
                 .limit(config.REBUILD_BATCH_SIZE)
                 .sort('-_id')
                 .lean()
                 .exec());
-            // return P.resolve(md_store.DataChunk.collection.find(query, {
-            //     limit: config.REBUILD_BATCH_SIZE,
-            //     sort: {
-            //         _id: -1
-            //     }
-            // }).toArray());
         })
-        .then(function(chunks) {
+        .then(function(chunk_ids) {
 
             // update the last_chunk_id for next time
-            if (chunks.length === config.REBUILD_BATCH_SIZE) {
-                self.last_chunk_id = chunks[0]._id;
+            if (chunk_ids.length === config.REBUILD_BATCH_SIZE) {
+                self.last_chunk_id = chunk_ids[0];
             } else {
                 self.last_chunk_id = undefined;
             }
 
-            if (chunks.length) {
-                dbg.log0('SCRUBBER:', 'WORKING ON', chunks.length, 'CHUNKS');
-                let builder = new map_builder.MapBuilder(chunks);
+            if (chunk_ids.length) {
+                dbg.log0('SCRUBBER:', 'WORKING ON', chunk_ids.length, 'CHUNKS');
+                let builder = new map_builder.MapBuilder(chunk_ids);
                 return builder.run()
                     .catch(function(err) {
                         dbg.error('SCRUBBER:', 'BUILD ERROR', err, err.stack);

src/server/node_services/nodes_monitor.js

@@ -1556,7 +1556,7 @@ class NodesMonitor extends EventEmitter {
                 // we update the stage marker even if failed to advance the scan
                 act.stage.marker = res.marker;
                 blocks_size = res.blocks_size;
-                const builder = new MapBuilder(res.chunks);
+                const builder = new MapBuilder(res.chunk_ids);
                 return builder.run();
             })
             .then(() => {

src/server/object_services/map_builder.js

@@ -13,11 +13,11 @@ const map_deleter = require('./map_deleter');
 const mongo_utils = require('../../util/mongo_utils');
 const system_store = require('../system_services/system_store').get_instance();
 const node_allocator = require('../node_services/node_allocator');
-const system_server_utils = require('../utils/system_server_utils');
-// const promise_utils = require('../../util/promise_utils');
+const KeysLock = require('../../util/keys_lock');
 
 
 const replicate_block_sem = new Semaphore(config.IO_REPLICATE_CONCURRENCY);
+const builder_lock = new KeysLock();
 
 
 /**
@@ -31,35 +31,54 @@ const replicate_block_sem = new Semaphore(config.IO_REPLICATE_CONCURRENCY);
  */
 class MapBuilder {
 
-    constructor(chunks) {
-        this.chunks = chunks;
-        this.system_id = chunks[0] && chunks[0].system;
+    constructor(chunk_ids) {
+        this.chunk_ids = chunk_ids;
     }
 
     run() {
-        dbg.log1('MapBuilder.run:', 'batch start', this.chunks.length, 'chunks');
-        if (!this.chunks.length) return;
-        if (system_server_utils.system_in_maintenance(this.system_id)) return;
-        return P.resolve()
-            .then(() => P.join(
-                system_store.refresh(),
-                md_store.load_blocks_for_chunks(this.chunks),
-                md_store.load_parts_objects_for_chunks(this.chunks),
-                this.mark_building()
-            ))
-            .then(() => this.refresh_alloc())
-            .then(() => this.analyze_chunks())
-            // .then(() => this.refresh_alloc())
-            .then(() => this.allocate_blocks())
-            .then(() => this.replicate_blocks())
-            .then(() => this.update_db())
-            .then(() => {
-                // return error from the promise if any replication failed,
-                // so that caller will know the build isn't really complete,
-                // although it might partially succeeded
-                if (this.had_errors) {
-                    throw new Error('MapBuilder had errors');
-                }
+        dbg.log1('MapBuilder.run:', 'batch start', this.chunk_ids.length, 'chunks');
+        if (!this.chunk_ids.length) return;
+
+        return builder_lock.surround_keys(_.map(this.chunk_ids, String), () => {
+            return P.resolve(this.reload_chunks(this.chunk_ids))
+                .then(() => P.join(
+                    system_store.refresh(),
+                    md_store.load_blocks_for_chunks(this.chunks),
+                    md_store.load_parts_objects_for_chunks(this.chunks),
+                    this.mark_building()
+                ))
+                .then(() => this.refresh_alloc())
+                .then(() => this.analyze_chunks())
+                .then(() => this.allocate_blocks())
+                .then(() => this.replicate_blocks())
+                .then(() => this.update_db())
+                .then(() => {
+                    // return error from the promise if any replication failed,
+                    // so that caller will know the build isn't really complete,
+                    // although it might partially succeeded
+                    if (this.had_errors) {
+                        throw new Error('MapBuilder had errors');
+                    }
+                });
+        });
+    }
+
+
+    // In order to get the most relevant data regarding the chunks
+    // Note that there is always a possibility that the chunks will cease to exist
+    reload_chunks(chunk_ids) {
+        var query = {
+            _id: {
+                $in: chunk_ids
+            },
+            deleted: null
+        };
+
+        return P.resolve(md_store.DataChunk.find(query)
+                .lean()
+                .exec())
+            .then(chunks => {
+                this.chunks = chunks;
             });
     }
 

src/server/object_services/md_store.js

@@ -186,9 +186,13 @@ function iterate_node_chunks(system_id, node_id, marker, limit) {
             _id: {
                 $in: mongo_utils.uniq_ids(blocks, 'chunk')
             }
+        }, {
+            fields: {
+                _id: 1,
+            },
         }).toArray())
-        .then(chunks => ({
-            chunks: chunks,
+        .then(chunk_ids => ({
+            chunk_ids: chunk_ids,
             marker: blocks.length ? blocks[blocks.length - 1]._id : null,
             blocks_size: _.sumBy(blocks, 'size'),
         }));

src/test/unit_tests/index.js

@@ -19,6 +19,7 @@ require('./test_s3_list_objects');
 // require('./test_debug_module');
 require('./test_job_queue');
 require('./test_linked_list');
+require('./test_keys_lock');
 require('./test_lru');
 require('./test_mongoose_logger');
 require('./test_prefetch');