Add a recovery option for the new store: skip_corrupted_meta_entries

src/blockstore/blockstore_heap.cpp

@@ -307,6 +307,7 @@ corrupted_block:
                 if (allow_corrupted)
                 {
                     fprintf(stderr, "Metadata block is corrupted, skipping\n");
+                    recheck_modified_blocks.insert(block_num);
                     break;
                 }
                 else
@@ -357,6 +358,7 @@ corrupted_object:
                 if (allow_corrupted)
                 {
                     fprintf(stderr, "Entry is corrupted, skipping\n");
+                    recheck_modified_blocks.insert(block_num);
                     block_offset += wr->size;
                     continue;
                 }

src/blockstore/blockstore_impl.h

@@ -78,6 +78,7 @@ public:
     // Suitable only for server SSDs with capacitors, requires disabled data and journal fsyncs
     int immediate_commit = IMMEDIATE_NONE;
     bool inmemory_meta = false;
+    bool skip_corrupted_meta_entries = false;
     uint32_t meta_write_recheck_parallelism = 0;
     // Maximum and minimum flusher count
     unsigned max_flusher_count = 0, min_flusher_count = 0;

src/blockstore/blockstore_init.cpp

@@ -225,7 +225,7 @@ resume_4:
         {
             // Handle result
             uint64_t loaded = 0;
-            int r = bs->heap->load_blocks(bufs[i].offset-bs->dsk.meta_block_size, bufs[i].size, bufs[i].buf, false, loaded);
+            int r = bs->heap->load_blocks(bufs[i].offset-bs->dsk.meta_block_size, bufs[i].size, bufs[i].buf, bs->skip_corrupted_meta_entries, loaded);
             if (r != 0)
                 exit(1);
             entries_loaded += loaded;

src/blockstore/blockstore_open.cpp

@@ -28,6 +28,7 @@ void blockstore_impl_t::parse_config(blockstore_config_t & config, bool init)
     throttle_target_parallelism = strtoull(config["throttle_target_parallelism"].c_str(), NULL, 10);
     throttle_threshold_us = strtoull(config["throttle_threshold_us"].c_str(), NULL, 10);
     perfect_csum_update = config["perfect_csum_update"] == "true" || config["perfect_csum_update"] == "1" || config["perfect_csum_update"] == "yes";
+    skip_corrupted_meta_entries = config["skip_corrupted_meta_entries"] == "true" || config["skip_corrupted_meta_entries"] == "1" || config["skip_corrupted_meta_entries"] == "yes";
     if (config["autosync_writes"] != "")
     {
         autosync_writes = strtoull(config["autosync_writes"].c_str(), NULL, 10);