diff --git a/libraries/chain/block_log.cpp b/libraries/chain/block_log.cpp
index 79234053aba..4829cfaaef9 100644
--- a/libraries/chain/block_log.cpp
+++ b/libraries/chain/block_log.cpp
@@ -7,6 +7,7 @@
 #include <boost/filesystem.hpp>
 #include <variant>
 #include <algorithm>
+#include <regex>
 
 namespace eosio { namespace chain {
 
@@ -63,7 +64,7 @@ namespace eosio { namespace chain {
             chain_context.emplace<genesis_state>();
             fc::raw::unpack(ds, std::get<genesis_state>(chain_context));
          } else if (block_log::contains_chain_id(version, first_block_num)) {
-            chain_context = chain_id_type{}; 
+            chain_context = chain_id_type{};
             ds >> std::get<chain_id_type>(chain_context);
          } else {
             EOS_THROW(block_log_exception,
@@ -184,6 +185,12 @@ namespace eosio { namespace chain {
 
       using log_entry = std::variant<log_entry_v4, signed_block_v0>;
 
+      const block_header& get_block_header(const log_entry& entry) {
+         return std::visit(overloaded{[](const signed_block_v0& v) -> const block_header& { return v; },
+                                      [](const log_entry_v4& v) -> const block_header& { return v.block; }},
+                           entry);
+      }
+
       template <typename Stream>
       void unpack(Stream& ds, log_entry& entry) {
          std::visit(
@@ -191,54 +198,6 @@ namespace eosio { namespace chain {
                         [&ds](log_entry_v4& v) { unpack(ds, v); }},
              entry);
       }
-   } // namespace
-
-  namespace detail {
-
-     /**
-      * The implementation detail for the read/write access to the block log/index 
-      *
-      * @note All the non-static member functions require to fullfill the class invariant after execution unless exceptions are thrown.
-      * @invariant block_file.is_open() && index_file.is_open()
-      **/
-      class block_log_impl {
-         public:
-            signed_block_ptr   head;
-            fc::cfile          block_file;
-            fc::cfile          index_file;
-            bool               genesis_written_to_block_log = false;
-            block_log_preamble preamble;
-            static uint32_t    default_version;
-
-            block_log_impl(const fc::path& data_dir);
-
-            static void ensure_file_exists(fc::cfile& f) {
-               if (fc::exists(f.get_file_path())) return; 
-               f.open(fc::cfile::create_or_update_rw_mode);
-               f.close();
-            }
-
-            uint64_t get_block_pos(uint32_t block_num);
-
-            void reset(uint32_t first_block_num, std::variant<genesis_state, chain_id_type>&& chain_context);
-
-            void flush();
-
-            uint64_t append(const signed_block_ptr& b, packed_transaction::cf_compression_type segment_compression);
-
-            uint64_t write_log_entry(const signed_block& b,
-                                     packed_transaction::cf_compression_type segment_compression);
-
-
-            void                          read_block_header(block_header& bh, uint64_t file_pos);
-            std::unique_ptr<signed_block> read_block(uint64_t pos);
-            void                          read_head();            
-      };
-
-      uint32_t block_log_impl::default_version = block_log::max_supported_version;
-   } // namespace detail
-
-   namespace {
 
    void create_mapped_file(boost::iostreams::mapped_file_sink& sink, const std::string& path, uint64_t size) {
       using namespace boost::iostreams;
@@ -262,6 +221,7 @@ namespace eosio { namespace chain {
       }
 
       void close() { index.close(); }
+
     private:
       std::ptrdiff_t                     current_offset = 0;
       boost::iostreams::mapped_file_sink index;
@@ -271,24 +231,68 @@ namespace eosio { namespace chain {
       std::exception_ptr inner;
    };
 
+   template <typename Stream>
+   std::unique_ptr<signed_block> read_block(Stream&& ds, uint32_t version, uint32_t expect_block_num = 0) {
+      std::unique_ptr<signed_block> block;
+      if (version >= pruned_transaction_version) {
+         block = std::make_unique<signed_block>();
+         unpack(ds, *block);
+      } else {
+         signed_block_v0 block_v0;
+         fc::raw::unpack(ds, block_v0);
+         block = std::make_unique<signed_block>(std::move(block_v0), true);
+      }
+
+      if (expect_block_num != 0) {
+         EOS_ASSERT(!!block && block->block_num() == expect_block_num, block_log_exception,
+                     "Wrong block was read from block log.");
+      }
+
+      return block;
+   }
+
+   template <typename Stream>
+   block_id_type read_block_id(Stream&& ds, uint32_t version, uint32_t expect_block_num) {
+      if (version >= pruned_transaction_version) {
+         uint32_t size;
+         uint8_t  compression;
+         fc::raw::unpack(ds, size);
+         fc::raw::unpack(ds, compression);
+         EOS_ASSERT(compression == static_cast<uint8_t>(packed_transaction::cf_compression_type::none),
+                     block_log_exception, "Only \"none\" compression type is supported.");
+      }
+      block_header bh;
+      fc::raw::unpack(ds, bh);
+
+      EOS_ASSERT(bh.block_num() == expect_block_num, block_log_exception,
+                  "Wrong block header was read from block log.",
+                  ("returned", bh.block_num())("expected", expect_block_num));
+
+      return bh.calculate_id();
+   }
+
    /// Provide the read only view of the blocks.log file
    class block_log_data {
       boost::iostreams::mapped_file_source file;
       block_log_preamble                   preamble;
       uint64_t                             first_block_pos = block_log::npos;
-    public:
+   public:
       block_log_data() = default;
       block_log_data(const fc::path& path) { open(path); }
 
       const block_log_preamble& get_preamble() const { return preamble; }
 
       fc::datastream<const char*> open(const fc::path& path) {
+         if (file.is_open())
+            file.close();
          file.open(path.generic_string());
          fc::datastream<const char*> ds(file.data(), file.size());
          preamble.read_from(ds);
          first_block_pos = ds.tellp();
          return ds;
       }
+      
+      bool is_open() const { return file.is_open(); }
 
       const char*   data() const { return file.data(); }
       uint64_t      size() const { return file.size(); }
@@ -313,6 +317,8 @@ namespace eosio { namespace chain {
          //   block_id_type        previous;                   //bytes 14:45, low 4 bytes is big endian block number of
          //   previous block
 
+         EOS_ASSERT(position <= size(), block_log_exception, "Invalid block position ${position}", ("position", position));
+
          int blknum_offset = 14;
          blknum_offset += offset_to_block_start(version());
          uint32_t prev_block_num = read_buffer<uint32_t>(data() + position + blknum_offset);
@@ -325,8 +331,12 @@ namespace eosio { namespace chain {
          return last_block_num() - first_block_num() + 1;
       }
 
+      fc::datastream<const char*> datastream_at(uint64_t pos) {
+         return fc::datastream<const char*>(data() + pos, file.size() - pos);
+      }
+
       /**
-       *  Validate a block log entry WITHOUT deserializing the entire block data. 
+       *  Validate a block log entry WITHOUT deserializing the entire block data.
        **/
       void light_validate_block_entry_at(uint64_t pos, uint32_t expected_block_num) const {
          const uint32_t actual_block_num = block_num_at(pos);
@@ -358,10 +368,7 @@ namespace eosio { namespace chain {
             throw bad_block_excpetion{std::current_exception()};
          }
 
-         const block_header& header =
-             std::visit(overloaded{[](const signed_block_v0& v) -> const block_header& { return v; },
-                                   [](const log_entry_v4& v) -> const block_header& { return v.block; }},
-                        entry);
+         const block_header& header = get_block_header(entry);
 
          auto                id        = header.calculate_id();
          auto                block_num = block_header::num_from_id(id);
@@ -397,11 +404,15 @@ namespace eosio { namespace chain {
       block_log_index(const fc::path& path) { open(path); }
 
       void open(const fc::path& path) { 
+         if (file.is_open())
+            file.close();
          file.open(path.generic_string());
          EOS_ASSERT(file.size() % sizeof(uint64_t) == 0, block_log_exception, "The size of ${file} is not the multiple of sizeof(uint64_t)",
             ("file", path.generic_string()));
       }
 
+      bool is_open() const { return file.is_open(); }
+
       using iterator = const uint64_t*;
       iterator begin() const { return reinterpret_cast<iterator>(file.data()); }
       iterator end() const { return reinterpret_cast<iterator>(file.data() + file.size()); }
@@ -477,22 +488,247 @@ namespace eosio { namespace chain {
                                                                                      uint64_t first_block_position) {
       return reverse_block_position_iterator<BlockLogData>(t, first_block_position);
    }
+
+   template <typename Lambda>
+   void for_each_file_in_dir_matches(const fc::path& dir, const char* pattern, Lambda&& lambda) {
+      const std::regex my_filter(pattern);
+      std::smatch      what;
+      boost::filesystem::directory_iterator end_itr; // Default ctor yields past-the-end
+      for (boost::filesystem::directory_iterator p(dir); p != end_itr; ++p) {
+         // Skip if not a file
+         if (!boost::filesystem::is_regular_file(p->status()))
+            continue;
+         // skip if it's not match blocks-*-*.log
+         if (!std::regex_match(p->path().filename().string(), what, my_filter))
+            continue;
+         lambda(p->path());
+      }
+   } 
    } // namespace
 
-   block_log::block_log(const fc::path& data_dir)
-       : my(new detail::block_log_impl(data_dir)) {
-   }
+   struct block_log_catalog {
+      using block_num_t = uint32_t;
+
+      struct mapped_type {
+         block_num_t             last_block_num;
+         boost::filesystem::path filename_base;
+      };
+      using collection_t              = boost::container::flat_map<block_num_t, mapped_type>;
+      using size_type                 = collection_t::size_type;
+      static constexpr size_type npos = std::numeric_limits<size_type>::max();
+
+      boost::filesystem::path    archive_dir;
+      size_type                  max_retained_files = 10;
+      collection_t               collection;
+      size_type                  active_index = npos;
+      block_log_data             log_data;
+      block_log_index            log_index;
+      chain_id_type              chain_id;
+
+      bool empty() const { return collection.empty();  }
+
+      void open(fc::path block_dir) {
+         for_each_file_in_dir_matches(block_dir, R"(blocks-\d+-\d+\.log)", [this](boost::filesystem::path path) {
+            auto log_path               = path;
+            auto index_path             = path.replace_extension("index");
+            auto path_without_extension = log_path.parent_path() / log_path.stem().string();
+
+            block_log_data log(log_path);
+
+            if (chain_id.empty()) {
+               chain_id = log.chain_id();
+            } else {
+               EOS_ASSERT(chain_id == log.chain_id(), block_log_exception, "block log file ${path} has a different chain id",
+                           ("path", log_path.generic_string()));
+            }
+
+            // check if index file matches the log file
+            if (!index_matches_data(index_path, log))
+               block_log::construct_index(log_path, index_path);
+
+            auto existing_itr = collection.find(log.first_block_num());
+            if (existing_itr != collection.end()) {
+               if (log.last_block_num() <= existing_itr->second.last_block_num) {
+                  wlog("${log_path} contains the overlapping range with ${existing_path}.log, droping ${log_path} "
+                       "from catelog",
+                       ("log_path", log_path.string())("existing_path", existing_itr->second.filename_base.string()));
+                  return;
+               }
+               else {
+                  wlog("${log_path} contains the overlapping range with ${existing_path}.log, droping ${existing_path}.log "
+                       "from catelog",
+                       ("log_path", log_path.string())("existing_path", existing_itr->second.filename_base.string()));
+               }
+            }
+
+            collection.insert_or_assign(log.first_block_num(), mapped_type{ log.last_block_num(), path_without_extension });
+         });
+      }  
+
+      bool index_matches_data(const boost::filesystem::path& index_path, const block_log_data& log) const {
+         if (boost::filesystem::exists(index_path) &&
+             boost::filesystem::file_size(index_path) / sizeof(uint64_t) != log.num_blocks()) {
+            // make sure the last 8 bytes of index and log matches
+
+            fc::cfile index_file;
+            index_file.set_file_path(index_path);
+            index_file.open("r");
+            index_file.seek_end(-sizeof(uint64_t));
+            uint64_t pos;
+            index_file.read(reinterpret_cast<char*>(&pos), sizeof(pos));
+            return pos == log.last_block_position();
+         }
+         return false;
+      }
+
+      bool set_active_item(uint32_t block_num) {
+         try {
+            if (active_index != npos) {
+               auto active_item = collection.nth(active_index);
+               if (active_item->first <= block_num && block_num <= active_item->second.last_block_num) {
+                  if (!log_index.is_open()) {
+                     log_index.open(active_item->second.filename_base.replace_extension("index"));
+                  }
+                  return true;
+               }
+            }
+            if (collection.empty() || block_num < collection.begin()->first )
+               return false;
+
+            auto it = --collection.upper_bound(block_num);
+
+            if (block_num <= it->second.last_block_num ) {
+               auto name = it->second.filename_base;
+               log_data.open(name.replace_extension("log"));
+               log_index.open(name.replace_extension("index"));
+               this->active_index = collection.index_of(it);
+               return true;
+            }
+            return false;
+         } catch (...) {
+            this->active_index = npos;
+            return false;
+         }
+      }
+
+      std::pair<fc::datastream<const char*>, uint32_t> datastream_for_block(uint32_t block_num) {
+         auto pos = log_index.nth_block_position(block_num - log_data.first_block_num());
+         return std::make_pair(log_data.datastream_at(pos), log_data.get_preamble().version);
+      }
+
+      /// Add a new entry into the catalog.
+      ///
+      /// Notice that \c start_block_num must be monotonically increasing between the invocations of this function
+      /// so that the new entry would be inserted at the end of the flat_map; otherwise, \c active_index would be invalidated
+      /// and the mapping between the log data their block range would be wrong. This function is only used during 
+      /// the splitting of block log. Using this function for other purpose should make sure if the monotonically 
+      /// increasing block num guarantee can be met.
+      void add(uint32_t start_block_num, uint32_t end_block_num, fc::path filename_base) {
+         if (this->collection.size() >= max_retained_files) {
+            auto items_to_erase = max_retained_files > 0 ? this->collection.size() - max_retained_files + 1 : this->collection.size();
+            for (auto it = this->collection.begin(); it < this->collection.begin() + items_to_erase; ++it) {
+               auto name = it->second.filename_base;
+               if (archive_dir.empty()) {
+                  // delete the old files when no backup dir is specified
+                  boost::filesystem::remove(name.replace_extension("log"));
+                  boost::filesystem::remove(name.replace_extension("index"));
+               } else {
+                  // move the the backup dir
+                  auto new_name = archive_dir / boost::filesystem::path(name).filename();
+                  boost::filesystem::rename(name.replace_extension("log"), new_name.replace_extension("log"));
+                  boost::filesystem::rename(name.replace_extension("index"), new_name.replace_extension("index"));
+               }
+            }
+            this->collection.erase(this->collection.begin(), this->collection.begin() + items_to_erase);
+            this->active_index = this->active_index == npos  || this->active_index < items_to_erase ? npos : this->active_index - items_to_erase;
+         }
+         if (max_retained_files > 0)
+            this->collection.emplace(start_block_num, mapped_type{end_block_num, filename_base});
+      }
+   };
+
+   namespace detail {
+
+      /**
+       * The implementation detail for the read/write access to the block log/index
+       *
+       * @note All the non-static member functions require to fullfill the class invariant after execution unless
+       *exceptions are thrown.
+       * @invariant block_file.is_open() && index_file.is_open()
+       **/
+      class block_log_impl {
+       public:
+         signed_block_ptr          head;
+         block_log_catalog         catalog;
+         fc::datastream<fc::cfile> block_file;
+         fc::datastream<fc::cfile> index_file;
+         bool                      genesis_written_to_block_log = false;
+         block_log_preamble        preamble;
+         size_t                    stride = std::numeric_limits<size_t>::max();
+         static uint32_t           default_version;
+
+         block_log_impl(const fc::path& data_dir, fc::path archive_dir, uint64_t stride, uint16_t max_retained_files, bool fix_irreversible_blocks);
+
+         static void ensure_file_exists(fc::cfile& f) {
+            if (fc::exists(f.get_file_path()))
+               return;
+            f.open(fc::cfile::create_or_update_rw_mode);
+            f.close();
+         }
+
+         uint64_t get_block_pos(uint32_t block_num);
+
+         void reset(uint32_t first_block_num, std::variant<genesis_state, chain_id_type>&& chain_context);
+
+         void flush();
+
+         uint64_t append(const signed_block_ptr& b, packed_transaction::cf_compression_type segment_compression);
+
+         uint64_t write_log_entry(const signed_block& b, packed_transaction::cf_compression_type segment_compression);
+
+         void split_log();
+         bool recover_from_incomplete_block_head(block_log_data& log_data, block_log_index& index);
+
+         block_id_type                 read_block_id_by_num(uint32_t block_num);
+         std::unique_ptr<signed_block> read_block_by_num(uint32_t block_num);
+         void                          read_head();
+      };
+      uint32_t block_log_impl::default_version = block_log::max_supported_version;
+   } // namespace detail
+
+   block_log::block_log(const fc::path& data_dir, fc::path archive_dir, uint64_t stride,
+                        uint16_t max_retained_files, bool fix_irreversible_blocks)
+       : my(new detail::block_log_impl(data_dir, archive_dir, stride, max_retained_files, fix_irreversible_blocks)) {}
+
+   block_log::~block_log() {}
 
-   block_log::~block_log(){}
 
    void block_log::set_version(uint32_t ver) { detail::block_log_impl::default_version = ver; }
+   uint32_t block_log::version() const { return my->preamble.version; }
+
+   detail::block_log_impl::block_log_impl(const fc::path& data_dir, fc::path archive_dir, uint64_t stride,
+                                          uint16_t max_retained_files, bool fix_irreversible_blocks) {
 
-   detail::block_log_impl::block_log_impl(const fc::path& data_dir) {
       if (!fc::is_directory(data_dir))
          fc::create_directories(data_dir);
+      else
+         catalog.open(data_dir);
+
+      if (! archive_dir.empty()) {
+         if (archive_dir.is_relative())
+            archive_dir = data_dir / archive_dir;
+
+         if (!fc::is_directory(archive_dir))
+            fc::create_directories(archive_dir);
+      }
+      
+
+      catalog.archive_dir        = archive_dir;
+      catalog.max_retained_files = max_retained_files;
+      this->stride               = stride;
 
-      block_file.set_file_path( data_dir / "blocks.log" );
-      index_file.set_file_path( data_dir / "blocks.index" );
+      block_file.set_file_path(data_dir / "blocks.log");
+      index_file.set_file_path(data_dir / "blocks.index");
       /* On startup of the block log, there are several states the log file and the index file can be
        * in relation to each other.
        *
@@ -521,16 +757,34 @@ namespace eosio { namespace chain {
          block_log_data log_data(block_file.get_file_path());
          preamble = log_data.get_preamble();
 
+         EOS_ASSERT(catalog.chain_id.empty() || catalog.chain_id == preamble.chain_id(), block_log_exception,
+                    "block log file ${path} has a different chain id", ("path", block_file.get_file_path()));
+
          genesis_written_to_block_log = true; // Assume it was constructed properly.
 
          if (index_size) {
             ilog("Index is nonempty");
-            block_log_index index(index_file.get_file_path());
-            
-            if (log_data.last_block_position() != index.back()) {
-               ilog("The last block positions from blocks.log and blocks.index are different, Reconstructing index...");
-               block_log::construct_index(block_file.get_file_path(), index_file.get_file_path());
+            if (index_size % sizeof(uint64_t) == 0) {
+               block_log_index index(index_file.get_file_path());
+               
+               if (log_data.last_block_position() != index.back()) {      
+                  if (!fix_irreversible_blocks)  {
+                     ilog("The last block positions from blocks.log and blocks.index are different, Reconstructing index...");
+                     block_log::construct_index(block_file.get_file_path(), index_file.get_file_path());
+                  }        
+                  else if (!recover_from_incomplete_block_head(log_data, index)) {
+                     block_log::repair_log(block_file.get_file_path().parent_path(), UINT32_MAX);
+                     block_log::construct_index(block_file.get_file_path(), index_file.get_file_path());
+                  }
+               } else if (fix_irreversible_blocks) {
+                  ilog("Irreversible blocks was not corrupted.");
+               }
             }
+            else {
+               if (fix_irreversible_blocks)
+                  block_log::repair_log(block_file.get_file_path().parent_path(), UINT32_MAX);
+               block_log::construct_index(block_file.get_file_path(), index_file.get_file_path());
+            } 
          } else {
             ilog("Index is empty. Reconstructing index...");
             block_log::construct_index(block_file.get_file_path(), index_file.get_file_path());
@@ -583,12 +837,38 @@ namespace eosio { namespace chain {
                    ("expected", (b->block_num() - preamble.first_block_num) * sizeof(uint64_t)));
 
          auto pos = write_log_entry(*b, segment_compression);
-         head = b;
+         head     = b;
+         if (b->block_num() % stride == 0) {
+            split_log();
+         }
          return pos;
       }
       FC_LOG_AND_RETHROW()
    }
 
+   void detail::block_log_impl::split_log() { 
+      block_file.close();
+      index_file.close();
+
+      auto      data_dir = block_file.get_file_path().parent_path();
+      const int bufsize  = 64;
+      char      filename[bufsize];
+      int n = snprintf(filename, bufsize, "blocks-%d-%d", preamble.first_block_num, this->head->block_num());
+      catalog.add(preamble.first_block_num, this->head->block_num(), data_dir / filename);
+      strncpy(filename + n, ".log", bufsize-n);
+      boost::filesystem::rename(block_file.get_file_path(), data_dir / filename);
+      strncpy(filename + n, ".index", bufsize-n);
+      boost::filesystem::rename(index_file.get_file_path(), data_dir / filename);
+      
+      block_file.open(fc::cfile::truncate_rw_mode);
+      index_file.open(fc::cfile::truncate_rw_mode);
+      preamble.version         = block_log::max_supported_version;
+      preamble.chain_context   = preamble.chain_id();
+      preamble.first_block_num = this->head->block_num() + 1;
+      preamble.write_to(block_file);
+      flush();
+   }
+
    void detail::block_log_impl::flush() {
       block_file.flush();
       index_file.flush();
@@ -614,67 +894,47 @@ namespace eosio { namespace chain {
       append(first_block, segment_compression);
    }
 
-   void block_log::reset( const chain_id_type& chain_id, uint32_t first_block_num ) {
-      EOS_ASSERT( first_block_num > 1, block_log_exception,
-                  "Block log version ${ver} needs to be created with a genesis state if starting from block number 1." );
+   void block_log::reset(const chain_id_type& chain_id, uint32_t first_block_num) {
+      EOS_ASSERT(first_block_num > 1, block_log_exception,
+                 "Block log version ${ver} needs to be created with a genesis state if starting from block number 1.");
+
+      EOS_ASSERT(my->catalog.chain_id.empty() || chain_id == my->catalog.chain_id, block_log_exception,
+                 "Trying to reset to the chain to a different chain id");
+
       my->reset(first_block_num, chain_id);
       my->head.reset();
    }
 
-   std::unique_ptr<signed_block> detail::block_log_impl::read_block(uint64_t pos) {
-      block_file.seek(pos);
-      auto ds = block_file.create_datastream();
-      if (preamble.version >= pruned_transaction_version) {
-         auto block = std::make_unique<signed_block>();
-         unpack(ds, *block);
-         return block;
-      } else {
-         signed_block_v0 block;
-         fc::raw::unpack(ds, block);
-         return std::make_unique<signed_block>(std::move(block), true);
+   std::unique_ptr<signed_block> detail::block_log_impl::read_block_by_num(uint32_t block_num) {
+      uint64_t pos = get_block_pos(block_num);
+      if (pos != block_log::npos) {
+         block_file.seek(pos);
+         return read_block(block_file, preamble.version, block_num);
+      } else if (catalog.set_active_item(block_num)) {
+         auto [ds, version] = catalog.datastream_for_block(block_num);
+         return read_block(ds, version, block_num);
       }
+      return {};
    }
 
-   void detail::block_log_impl::read_block_header(block_header& bh, uint64_t pos) {
-      block_file.seek(pos);
-      auto ds = block_file.create_datastream();
-
-      if (preamble.version >= pruned_transaction_version ) {
-         uint32_t size;
-         uint8_t  compression;
-         fc::raw::unpack(ds, size);
-         fc::raw::unpack(ds, compression);
-         EOS_ASSERT( compression == static_cast<uint8_t>(packed_transaction::cf_compression_type::none), block_log_exception ,
-                     "Only \"none\" compression type is supported.");
+   block_id_type detail::block_log_impl::read_block_id_by_num(uint32_t block_num) {
+      uint64_t pos = get_block_pos(block_num);
+      if (pos != block_log::npos) {
+         block_file.seek(pos);
+         return read_block_id(block_file, preamble.version, block_num);
+      } else if (catalog.set_active_item(block_num)) {
+         auto [ds, version] = catalog.datastream_for_block(block_num);
+         return read_block_id(ds, version, block_num);
       }
-      fc::raw::unpack(ds, bh);
+      return {};
    }
 
    std::unique_ptr<signed_block> block_log::read_signed_block_by_num(uint32_t block_num) const {
-      try {
-         std::unique_ptr<signed_block> b;
-         uint64_t pos = my->get_block_pos(block_num);
-         if (pos != npos) {
-            b = my->read_block(pos);
-            EOS_ASSERT(b->block_num() == block_num, block_log_exception,
-                      "Wrong block was read from block log.");
-         }
-         return b;
-      } FC_LOG_AND_RETHROW()
+      return my->read_block_by_num(block_num);
    }
 
    block_id_type block_log::read_block_id_by_num(uint32_t block_num) const {
-      try {
-         uint64_t pos = my->get_block_pos(block_num);
-         if (pos != npos) {
-            block_header bh;
-            my->read_block_header(bh, pos);
-            EOS_ASSERT(bh.block_num() == block_num, block_log_exception,
-                       "Wrong block header was read from block log.", ("returned", bh.block_num())("expected", block_num));
-            return bh.calculate_id();
-         }
-         return {};
-      } FC_LOG_AND_RETHROW()
+      return my->read_block_id_by_num(block_num);
    }
 
    uint64_t detail::block_log_impl::get_block_pos(uint32_t block_num) {
@@ -692,8 +952,9 @@ namespace eosio { namespace chain {
       block_file.seek_end(-sizeof(pos));
       block_file.read((char*)&pos, sizeof(pos));
       if (pos != block_log::npos) {
-         head = read_block(pos);
-      } 
+         block_file.seek(pos);
+         head = read_block(block_file, preamble.version);
+      }
    }
 
    const signed_block_ptr& block_log::head() const {
@@ -701,6 +962,8 @@ namespace eosio { namespace chain {
    }
 
    uint32_t block_log::first_block_num() const {
+      if (!my->catalog.empty())
+         return my->catalog.collection.begin()->first;
       return my->preamble.first_block_num;
    }
 
@@ -749,7 +1012,43 @@ namespace eosio { namespace chain {
       
    }
 
-   fc::path block_log::repair_log(const fc::path& data_dir, uint32_t truncate_at_block) {
+   bool detail::block_log_impl::recover_from_incomplete_block_head(block_log_data& log_data, block_log_index& index) {
+      const uint64_t pos = index.back();
+      if (log_data.size() <= pos) {
+         // index refers to an invalid position, we cannot recover from it 
+         return false;
+      }
+
+      log_entry entry;
+      if (preamble.version < pruned_transaction_version) {
+         entry.emplace<signed_block_v0>();
+      }
+
+      const uint32_t expected_block_num = log_data.first_block_num() + index.num_blocks() - 1;
+      fc::datastream<const char*> ds(log_data.data() + pos, log_data.size() - pos);
+
+      try {
+         unpack(ds, entry);
+         const block_header& header = get_block_header(entry);
+         if (header.block_num() != expected_block_num) {
+            return false;
+         }
+         uint64_t tmp_pos = std::numeric_limits<uint64_t>::max();
+         ds.read(reinterpret_cast<char*>(&tmp_pos), sizeof(tmp_pos));
+         if (tmp_pos != pos)
+            return false;
+
+         const auto size_to_trim            = ds.remaining();
+         const auto trimmed_block_file_size = log_data.size() - size_to_trim;
+
+         write_incomplete_block_data(block_file.get_file_path().parent_path(), fc::time_point::now(), expected_block_num + 1,
+                                    log_data.data() + trimmed_block_file_size, size_to_trim);
+         boost::filesystem::resize_file(block_file.get_file_path(), trimmed_block_file_size);
+         return true;
+      } catch (...) { return false; }
+   }
+
+   fc::path block_log::repair_log(const fc::path& data_dir, uint32_t truncate_at_block, const char* reversible_block_dir_name) {
       ilog("Recovering Block Log...");
       EOS_ASSERT(fc::is_directory(data_dir) && fc::is_regular_file(data_dir / "blocks.log"), block_log_not_found,
                  "Block log not found in '${blocks_dir}'", ("blocks_dir", data_dir));
@@ -767,10 +1066,16 @@ namespace eosio { namespace chain {
                  "Cannot move existing blocks directory to already existing directory '${new_blocks_dir}'",
                  ("new_blocks_dir", backup_dir));
 
-      fc::rename(blocks_dir, backup_dir);
+      fc::create_directories(backup_dir);
+      fc::rename(blocks_dir / "blocks.log", backup_dir / "blocks.log");
+      if (fc::exists(blocks_dir/ "blocks.index")) {
+         fc::rename(blocks_dir/ "blocks.index", backup_dir/ "blocks.index");
+      }
+      if (strlen(reversible_block_dir_name) && fc::is_directory(blocks_dir/reversible_block_dir_name)) {
+         fc::rename(blocks_dir/ reversible_block_dir_name, backup_dir/ reversible_block_dir_name);
+      }
       ilog("Moved existing blocks directory to backup location: '${new_blocks_dir}'", ("new_blocks_dir", backup_dir));
 
-      fc::create_directories(blocks_dir);
       const auto block_log_path  = blocks_dir / "blocks.log";
       const auto block_file_name = block_log_path.generic_string();
 
@@ -827,8 +1132,10 @@ namespace eosio { namespace chain {
       return backup_dir;
    }
 
-   fc::optional<genesis_state> block_log::extract_genesis_state( const fc::path& data_dir ) {
-      return block_log_data(data_dir / "blocks.log").get_genesis_state();
+   fc::optional<genesis_state> block_log::extract_genesis_state( const fc::path& block_dir ) {
+      boost::filesystem::path p(block_dir / "blocks.log");
+      for_each_file_in_dir_matches(block_dir, R"(blocks-1-\d+\.log)", [&p](boost::filesystem::path log_path) { p = log_path; });
+      return block_log_data(p).get_genesis_state();
    }
       
    chain_id_type block_log::extract_chain_id( const fc::path& data_dir ) {
diff --git a/libraries/chain/controller.cpp b/libraries/chain/controller.cpp
index a2812eb50b5..2b615ad5d3b 100644
--- a/libraries/chain/controller.cpp
+++ b/libraries/chain/controller.cpp
@@ -310,7 +310,7 @@ struct controller_impl {
     reversible_blocks( cfg.blocks_dir/config::reversible_blocks_dir_name,
         cfg.read_only ? database::read_only : database::read_write,
         cfg.reversible_cache_size, false, cfg.db_map_mode, cfg.db_hugepage_paths ),
-    blog( cfg.blocks_dir ),
+    blog( cfg.blocks_dir, cfg.blocks_archive_dir, cfg.blocks_log_stride, cfg.max_retained_block_files, cfg.fix_irreversible_blocks),
     fork_db( cfg.state_dir ),
     wasmif( cfg.wasm_runtime, cfg.eosvmoc_tierup, db, cfg.state_dir, cfg.eosvmoc_config ),
     resource_limits( db, [&s]() { return s.get_deep_mind_logger(); }),
diff --git a/libraries/chain/include/eosio/chain/block_log.hpp b/libraries/chain/include/eosio/chain/block_log.hpp
index 7a56922aafd..cd5d3ed34c7 100644
--- a/libraries/chain/include/eosio/chain/block_log.hpp
+++ b/libraries/chain/include/eosio/chain/block_log.hpp
@@ -8,7 +8,7 @@ namespace eosio { namespace chain {
    namespace detail { class block_log_impl; }
 
    /* The block log is an external append only log of the blocks with a header. Blocks should only
-    * be written to the log after they irreverisble as the log is append only. The log is a doubly
+    * be written to the log after they are irreversible as the log is append only. The log is a doubly
     * linked list of blocks. There is a secondary index file of only block positions that enables
     * O(1) random access lookup by block number.
     *
@@ -34,7 +34,8 @@ namespace eosio { namespace chain {
 
    class block_log {
       public:
-         block_log(const fc::path& data_dir);
+         block_log(const fc::path& data_dir, fc::path backup_dir = fc::path(), uint64_t stride=1000, 
+                   uint16_t max_retained_files=10, bool fix_irreversible_blocks=false);
          block_log(block_log&& other) = default;
          ~block_log();
          
@@ -64,7 +65,7 @@ namespace eosio { namespace chain {
          static const uint32_t min_supported_version;
          static const uint32_t max_supported_version;
 
-         static fc::path repair_log( const fc::path& data_dir, uint32_t truncate_at_block = UINT32_MAX );
+         static fc::path repair_log( const fc::path& data_dir, uint32_t truncate_at_block = UINT32_MAX, const char* reversible_block_dir_name="" );
 
          static fc::optional<genesis_state> extract_genesis_state( const fc::path& data_dir );
 
@@ -83,6 +84,7 @@ namespace eosio { namespace chain {
 
          // used for unit test to generate older version blocklog
          static void set_version(uint32_t);
+         uint32_t    version() const;
 
          /**
           * @param n Only test 1 block out of every n blocks. If n is 0, it is maximum between 1 and the ceiling of the total number blocks divided by 8.
diff --git a/libraries/chain/include/eosio/chain/chain_id_type.hpp b/libraries/chain/include/eosio/chain/chain_id_type.hpp
index 40521078449..fb165a38039 100644
--- a/libraries/chain/include/eosio/chain/chain_id_type.hpp
+++ b/libraries/chain/include/eosio/chain/chain_id_type.hpp
@@ -38,6 +38,8 @@ namespace chain {
 
       void reflector_init()const;
 
+      bool empty() const { return *this == chain_id_type{};}
+
       private:
          chain_id_type() = default;
 
@@ -51,6 +53,7 @@ namespace chain {
          friend struct eosio::handshake_message;
          friend class block_log;
          friend struct block_log_preamble;
+         friend struct block_log_catalog;
          friend class controller;
          friend struct controller_impl;
          friend class global_property_object;
diff --git a/libraries/chain/include/eosio/chain/config.hpp b/libraries/chain/include/eosio/chain/config.hpp
index ed5d915b5eb..f805f414219 100644
--- a/libraries/chain/include/eosio/chain/config.hpp
+++ b/libraries/chain/include/eosio/chain/config.hpp
@@ -9,7 +9,10 @@ namespace eosio { namespace chain { namespace config {
 typedef __uint128_t uint128_t;
 
 const static auto default_blocks_dir_name    = "blocks";
-const static auto reversible_blocks_dir_name = "reversible";
+const static auto default_blocks_archive_dir_name  = "archive";
+const static auto default_blocks_log_stride        = UINT32_MAX;
+const static auto default_max_retained_block_files = 10;
+const static auto reversible_blocks_dir_name       = "reversible";
 const static auto default_reversible_cache_size = 340*1024*1024ll;/// 1MB * 340 blocks based on 21 producer BFT delay
 const static auto default_reversible_guard_size = 2*1024*1024ll;/// 1MB * 340 blocks based on 21 producer BFT delay
 
diff --git a/libraries/chain/include/eosio/chain/controller.hpp b/libraries/chain/include/eosio/chain/controller.hpp
index 1ebf294a3bd..b3daa922076 100644
--- a/libraries/chain/include/eosio/chain/controller.hpp
+++ b/libraries/chain/include/eosio/chain/controller.hpp
@@ -66,6 +66,7 @@ namespace eosio { namespace chain {
             flat_set< pair<account_name, action_name> > action_blacklist;
             flat_set<public_key_type> key_blacklist;
             path                     blocks_dir                 = chain::config::default_blocks_dir_name;
+            path                     blocks_archive_dir         = chain::config::default_blocks_archive_dir_name;
             path                     state_dir                  = chain::config::default_state_dir_name;
             uint64_t                 state_size                 = chain::config::default_state_size;
             uint64_t                 state_guard_size           = chain::config::default_state_guard_size;
@@ -73,12 +74,15 @@ namespace eosio { namespace chain {
             uint64_t                 reversible_guard_size      = chain::config::default_reversible_guard_size;
             uint32_t                 sig_cpu_bill_pct           = chain::config::default_sig_cpu_bill_pct;
             uint16_t                 thread_pool_size           = chain::config::default_controller_thread_pool_size;
+            uint16_t                 max_retained_block_files   = chain::config::default_max_retained_block_files;
+            uint64_t                 blocks_log_stride          = chain::config::default_blocks_log_stride;
             fc::microseconds         abi_serializer_max_time_us = fc::microseconds(chain::config::default_abi_serializer_max_time_us);
             bool                     read_only                  = false;
             bool                     force_all_checks           = false;
             bool                     disable_replay_opts        = false;
             bool                     contracts_console          = false;
             bool                     allow_ram_billing_in_notify = false;
+            bool                     fix_irreversible_blocks    = false;
             uint32_t                 maximum_variable_signature_length = chain::config::default_max_variable_signature_length;
             bool                     disable_all_subjective_mitigations = false; //< for developer & testing purposes, can be configured using `disable-all-subjective-mitigations` when `EOSIO_DEVELOPER` build option is provided
             uint32_t                 terminate_at_block     = 0; //< primarily for testing purposes
diff --git a/libraries/fc/include/fc/filesystem.hpp b/libraries/fc/include/fc/filesystem.hpp
index 7484f2a5a2a..e3bfc3fa158 100644
--- a/libraries/fc/include/fc/filesystem.hpp
+++ b/libraries/fc/include/fc/filesystem.hpp
@@ -81,6 +81,7 @@ namespace fc {
 
       bool       is_relative()const;
       bool       is_absolute()const;
+      bool       empty() const;
 
       static char    separator_char;
 
diff --git a/libraries/fc/src/filesystem.cpp b/libraries/fc/src/filesystem.cpp
index 2337d101794..24a96c99a17 100644
--- a/libraries/fc/src/filesystem.cpp
+++ b/libraries/fc/src/filesystem.cpp
@@ -170,6 +170,8 @@ namespace fc {
   bool path::is_relative()const { return _p->is_relative(); }
   bool path::is_absolute()const { return _p->is_absolute(); }
 
+  bool path::empty() const { return _p->empty(); }
+
       directory_iterator::directory_iterator( const fc::path& p )
       :_p(p){}
 
diff --git a/plugins/chain_plugin/chain_plugin.cpp b/plugins/chain_plugin/chain_plugin.cpp
index 6092f037248..ca66643e9a5 100644
--- a/plugins/chain_plugin/chain_plugin.cpp
+++ b/plugins/chain_plugin/chain_plugin.cpp
@@ -150,7 +150,6 @@ class chain_plugin_impl {
 
 
    fc::optional<fork_database>      fork_db;
-   fc::optional<block_log>          block_logger;
    fc::optional<controller::config> chain_config;
    fc::optional<controller>         chain;
    fc::optional<genesis_state>      genesis;
@@ -228,6 +227,22 @@ void chain_plugin::set_program_options(options_description& cli, options_descrip
    cfg.add_options()
          ("blocks-dir", bpo::value<bfs::path>()->default_value("blocks"),
           "the location of the blocks directory (absolute path or relative to application data dir)")
+         ("blocks-log-stride", bpo::value<uint32_t>()->default_value(config::default_blocks_log_stride),
+         "split the block log file when the head block number is the multiple of the split factor\n"
+         "When the stride is reached, the current block log and index will be renamed 'blocks-<start num>-<end num>.log/index'\n"
+         "and a new current block log and index will be created with the most recent block. All files following\n"
+         "this format will be used to construct an extended block log.")
+         ("max-retained-block-files", bpo::value<uint16_t>()->default_value(config::default_max_retained_block_files),
+          "the maximum number of blocks files to retain so that the blocks in those files can be queried.\n" 
+          "When the number is reached, the oldest block file would be moved to archive dir or deleted if the archive dir is empty.\n"
+          "The retained block log files should not be manipulated by users." )
+         ("blocks-archive-dir", bpo::value<bfs::path>()->default_value(config::default_blocks_archive_dir_name),
+          "the location of the blocks archive directory (absolute path or relative to blocks dir).\n"
+          "If the value is empty, blocks files beyond the retained limit will be deleted.\n"
+          "All files in the archive directory are completely under user's control, i.e. they won't be accessed by nodeos anymore.")
+         ("fix-irreversible-blocks", bpo::value<bool>()->default_value("false"),
+          "When the existing block log is inconsistent with the index, allows fixing the block log and index files automatically - that is, " 
+          "it will take the highest indexed block if it is valid; otherwise it will repair the block log and reconstruct the index.")
          ("protocol-features-dir", bpo::value<bfs::path>()->default_value("protocol_features"),
           "the location of the protocol_features directory (absolute path or relative to application config dir)")
          ("checkpoint", bpo::value<vector<string>>()->composing(), "Pairs of [BLOCK_NUM,BLOCK_ID] that should be enforced as checkpoints.")
@@ -612,7 +627,7 @@ void
 chain_plugin::do_hard_replay(const variables_map& options) {
          ilog( "Hard replay requested: deleting state database" );
          clear_directory_contents( my->chain_config->state_dir );
-         auto backup_dir = block_log::repair_log( my->blocks_dir, options.at( "truncate-at-block" ).as<uint32_t>());
+         auto backup_dir = block_log::repair_log( my->blocks_dir, options.at( "truncate-at-block" ).as<uint32_t>(),config::reversible_blocks_dir_name);
          if( fc::exists( backup_dir / config::reversible_blocks_dir_name ) ||
              options.at( "fix-reversible-blocks" ).as<bool>()) {
             // Do not try to recover reversible blocks if the directory does not exist, unless the option was explicitly provided.
@@ -742,6 +757,10 @@ void chain_plugin::plugin_initialize(const variables_map& options) {
       my->chain_config->blocks_dir = my->blocks_dir;
       my->chain_config->state_dir = app().data_dir() / config::default_state_dir_name;
       my->chain_config->read_only = my->readonly;
+      my->chain_config->blocks_archive_dir = options.at("blocks-archive-dir").as<bfs::path>();
+      my->chain_config->blocks_log_stride  = options.at("blocks-log-stride").as<uint32_t>();
+      my->chain_config->max_retained_block_files = options.at("max-retained-block-files").as<uint16_t>();
+      my->chain_config->fix_irreversible_blocks = options.at("fix-irreversible-blocks").as<bool>();
 
       if (auto resmon_plugin = app().find_plugin<resource_monitor_plugin>()) {
         resmon_plugin->monitor_directory(my->chain_config->blocks_dir);
diff --git a/unittests/auth_tests.cpp b/unittests/auth_tests.cpp
index 87e360c10f1..5c2361bc04a 100644
--- a/unittests/auth_tests.cpp
+++ b/unittests/auth_tests.cpp
@@ -7,7 +7,6 @@
 #include <eosio/chain/resource_limits.hpp>
 #include <eosio/chain/resource_limits_private.hpp>
 
-#include <eosio/testing/tester_network.hpp>
 
 #ifdef NON_VALIDATING_TEST
 #define TESTER tester
diff --git a/unittests/delay_tests.cpp b/unittests/delay_tests.cpp
index 6313fdade7f..c0b0fb3bab4 100644
--- a/unittests/delay_tests.cpp
+++ b/unittests/delay_tests.cpp
@@ -1,6 +1,5 @@
 #include <eosio/chain/generated_transaction_object.hpp>
 #include <eosio/chain/global_property_object.hpp>
-#include <eosio/testing/tester_network.hpp>
 
 #include <boost/test/unit_test.hpp>
 
diff --git a/unittests/restart_chain_tests.cpp b/unittests/restart_chain_tests.cpp
index dd53b0c26a3..d5b1f5d45cf 100644
--- a/unittests/restart_chain_tests.cpp
+++ b/unittests/restart_chain_tests.cpp
@@ -11,6 +11,7 @@
 #include <contracts.hpp>
 #include <snapshots.hpp>
 #include <boost/iostreams/device/mapped_file.hpp>
+#include <fc/io/cfile.hpp>
 #include "test_cfd_transaction.hpp"
 
 using namespace eosio;
@@ -60,6 +61,7 @@ class replay_tester : public base_tester {
 struct restart_from_block_log_test_fixture {
    tester   chain;
    uint32_t cutoff_block_num;
+   bool     fix_irreversible_blocks = false;
    restart_from_block_log_test_fixture() {
       chain.create_account(N(replay1));
       chain.produce_blocks(1);
@@ -78,8 +80,9 @@ struct restart_from_block_log_test_fixture {
       chain.close();
    }
    ~restart_from_block_log_test_fixture() {
-      controller::config copied_config = chain.get_config();
-      auto               genesis       = chain::block_log::extract_genesis_state(chain.get_config().blocks_dir);
+      controller::config copied_config      = chain.get_config();
+      copied_config.fix_irreversible_blocks = this->fix_irreversible_blocks;
+      auto genesis                          = chain::block_log::extract_genesis_state(chain.get_config().blocks_dir);
       BOOST_REQUIRE(genesis);
 
       // remove the state files to make sure we are starting from block log
@@ -235,8 +238,232 @@ BOOST_FIXTURE_TEST_CASE(test_light_validation_restart_from_block_log_with_pruned
    BOOST_REQUIRE_NO_THROW(block_log::repair_log(blocks_dir));
 }
 
+BOOST_AUTO_TEST_CASE(test_split_log) {
+   namespace bfs = boost::filesystem;
+   fc::temp_directory temp_dir;
+
+   tester chain(
+         temp_dir,
+         [](controller::config& config) {
+            config.blocks_log_stride        = 20;
+            config.max_retained_block_files = 5;
+         },
+         true);
+   chain.produce_blocks(150);
+
+   auto blocks_dir = chain.get_config().blocks_dir;
+   auto blocks_archive_dir = chain.get_config().blocks_dir / chain.get_config().blocks_archive_dir;
+
+   BOOST_CHECK(bfs::exists( blocks_archive_dir / "blocks-1-20.log" ));
+   BOOST_CHECK(bfs::exists( blocks_archive_dir / "blocks-1-20.index" ));
+   BOOST_CHECK(bfs::exists( blocks_archive_dir / "blocks-21-40.log" ));
+   BOOST_CHECK(bfs::exists( blocks_archive_dir / "blocks-21-40.index" ));
+
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-41-60.log" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-41-60.index" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-61-80.log" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-61-80.index" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-81-100.log" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-81-100.index" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-101-120.log" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-101-120.index" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-121-140.log" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-121-140.index" ));
+
+   BOOST_CHECK( ! chain.control->fetch_block_by_number(40) );
+   
+   BOOST_CHECK( chain.control->fetch_block_by_number(81)->block_num() == 81 );
+   BOOST_CHECK( chain.control->fetch_block_by_number(90)->block_num() == 90 );
+   BOOST_CHECK( chain.control->fetch_block_by_number(100)->block_num() == 100 );
+
+   BOOST_CHECK( chain.control->fetch_block_by_number(41)->block_num() == 41 );
+   BOOST_CHECK( chain.control->fetch_block_by_number(50)->block_num() == 50 );
+   BOOST_CHECK( chain.control->fetch_block_by_number(60)->block_num() == 60 );
+
+   BOOST_CHECK( chain.control->fetch_block_by_number(121)->block_num() == 121 );
+   BOOST_CHECK( chain.control->fetch_block_by_number(130)->block_num() == 130 );
+   BOOST_CHECK( chain.control->fetch_block_by_number(140)->block_num() == 140 );
+
+   BOOST_CHECK( chain.control->fetch_block_by_number(145)->block_num() == 145);
+
+   BOOST_CHECK( ! chain.control->fetch_block_by_number(160));
+}
+
+BOOST_AUTO_TEST_CASE(test_split_log_no_archive) {
+
+   namespace bfs = boost::filesystem;
+   fc::temp_directory temp_dir;
+
+   tester chain(
+         temp_dir,
+         [](controller::config& config) {
+            config.blocks_archive_dir       = "";
+            config.blocks_log_stride        = 10;
+            config.max_retained_block_files = 5;
+         },
+         true);
+   chain.produce_blocks(75);
+
+   auto blocks_dir = chain.get_config().blocks_dir;
+   auto blocks_archive_dir = chain.get_config().blocks_dir / chain.get_config().blocks_archive_dir;
+
+   BOOST_CHECK(!bfs::exists( blocks_archive_dir / "blocks-1-10.log" ));
+   BOOST_CHECK(!bfs::exists( blocks_archive_dir / "blocks-1-10.index" ));
+   BOOST_CHECK(!bfs::exists( blocks_archive_dir / "blocks-11-20.log" ));
+   BOOST_CHECK(!bfs::exists( blocks_archive_dir / "blocks-11-20.index" ));
+   BOOST_CHECK(!bfs::exists( blocks_dir / "blocks-1-10.log" ));
+   BOOST_CHECK(!bfs::exists( blocks_dir / "blocks-1-10.index" ));
+   BOOST_CHECK(!bfs::exists( blocks_dir / "blocks-11-20.log" ));
+   BOOST_CHECK(!bfs::exists( blocks_dir / "blocks-11-20.index" ));
+
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-21-30.log" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-21-30.index" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-31-40.log" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-31-40.index" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-41-50.log" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-41-50.index" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-51-60.log" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-51-60.index" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-61-70.log" ));
+   BOOST_CHECK(bfs::exists( blocks_dir / "blocks-61-70.index" ));
+
+   BOOST_CHECK( ! chain.control->fetch_block_by_number(10));
+   BOOST_CHECK( chain.control->fetch_block_by_number(70));
+   BOOST_CHECK( ! chain.control->fetch_block_by_number(80));
+}
+
+void split_log_replay(uint32_t replay_max_retained_block_files) {
+   namespace bfs = boost::filesystem;
+   fc::temp_directory temp_dir;
+
+   const uint32_t stride = 20;
+
+   tester chain(
+         temp_dir,
+         [](controller::config& config) {
+            config.blocks_log_stride        = stride;
+            config.max_retained_block_files = 10;
+         },
+         true);
+   chain.produce_blocks(150);
+   
+   controller::config copied_config = chain.get_config();
+   auto               genesis       = chain::block_log::extract_genesis_state(chain.get_config().blocks_dir);
+   BOOST_REQUIRE(genesis);
+
+   chain.close();
+
+   // remove the state files to make sure we are starting from block log
+   remove_existing_states(copied_config);
+   // we need to remove the reversible blocks so that new blocks can be produced from the new chain
+   bfs::remove_all(copied_config.blocks_dir/"reversible");
+   copied_config.blocks_log_stride        = stride;
+   copied_config.max_retained_block_files = replay_max_retained_block_files;
+   tester from_block_log_chain(copied_config, *genesis);
+   BOOST_CHECK( from_block_log_chain.control->fetch_block_by_number(1)->block_num() == 1);
+   BOOST_CHECK( from_block_log_chain.control->fetch_block_by_number(75)->block_num() == 75);
+   BOOST_CHECK( from_block_log_chain.control->fetch_block_by_number(100)->block_num() == 100);
+   BOOST_CHECK( from_block_log_chain.control->fetch_block_by_number(150)->block_num() == 150);
+
+   // produce new blocks to cross the blocks_log_stride boundary
+   from_block_log_chain.produce_blocks(stride);
+
+   const auto previous_chunk_end_block_num = (from_block_log_chain.control->head_block_num() / stride) * stride;
+   const auto num_removed_blocks = std::min(stride * replay_max_retained_block_files, previous_chunk_end_block_num);
+   const auto min_retained_block_number = previous_chunk_end_block_num - num_removed_blocks + 1;
+
+   if (min_retained_block_number > 1) {
+      // old blocks beyond the max_retained_block_files will no longer be available
+      BOOST_CHECK(! from_block_log_chain.control->fetch_block_by_number(min_retained_block_number - 1));
+   }
+   BOOST_CHECK( from_block_log_chain.control->fetch_block_by_number(min_retained_block_number)->block_num() == min_retained_block_number);
+}
+
+BOOST_AUTO_TEST_CASE(test_split_log_replay_retained_block_files_10) {
+   split_log_replay(10);
+}
+
+BOOST_AUTO_TEST_CASE(test_split_log_replay_retained_block_files_5) {
+   split_log_replay(5);
+}
+
+BOOST_AUTO_TEST_CASE(test_split_log_replay_retained_block_files_1) {
+   split_log_replay(1);
+}
 
-void trim_blocklog_front() {
+BOOST_AUTO_TEST_CASE(test_split_log_replay_retained_block_files_0) {
+   split_log_replay(0);
+}
+
+BOOST_FIXTURE_TEST_CASE(auto_fix_with_incomplete_head,restart_from_block_log_test_fixture) {
+   auto& config = chain.get_config();
+   auto blocks_path = config.blocks_dir;
+   // write a few random bytes to block log indicating the last block entry is incomplete
+   fc::cfile logfile;
+   logfile.set_file_path(config.blocks_dir / "blocks.log");
+   logfile.open("ab");
+   const char random_data[] = "12345678901231876983271649837";
+   logfile.write(random_data, sizeof(random_data));
+   fix_irreversible_blocks = true;
+}
+
+BOOST_FIXTURE_TEST_CASE(auto_fix_with_corrupted_index,restart_from_block_log_test_fixture) {
+   auto& config = chain.get_config();
+   auto blocks_path = config.blocks_dir;
+   // write a few random index to block log indicating the index is corrupted
+   fc::cfile indexfile;
+   indexfile.set_file_path(config.blocks_dir / "blocks.index");
+   indexfile.open("ab");
+   uint64_t data = UINT64_MAX;
+   indexfile.write(reinterpret_cast<const char*>(&data), sizeof(data));
+   fix_irreversible_blocks = true;
+}
+
+BOOST_FIXTURE_TEST_CASE(auto_fix_with_corrupted_log_and_index,restart_from_block_log_test_fixture) {
+   auto& config = chain.get_config();
+   auto blocks_path = config.blocks_dir;
+   // write a few random bytes to block log and index
+   fc::cfile indexfile;
+   indexfile.set_file_path(config.blocks_dir / "blocks.index");
+   indexfile.open("ab");
+   const char random_index[] = "1234";
+   indexfile.write(reinterpret_cast<const char*>(&random_index), sizeof(random_index));
+
+   fc::cfile logfile;
+   logfile.set_file_path(config.blocks_dir / "blocks.log");
+   logfile.open("ab");
+   const char random_data[] = "12345678901231876983271649837";
+   logfile.write(random_data, sizeof(random_data));
+
+   fix_irreversible_blocks = true;
+}
+
+struct blocklog_version_setter {
+   blocklog_version_setter(uint32_t ver) { block_log::set_version(ver); };
+   ~blocklog_version_setter() { block_log::set_version(block_log::max_supported_version); };
+};
+
+BOOST_AUTO_TEST_CASE(test_split_from_v1_log) {
+   namespace bfs = boost::filesystem;
+   fc::temp_directory temp_dir;
+   blocklog_version_setter set_version(1); 
+   tester chain(
+         temp_dir,
+         [](controller::config& config) {
+            config.blocks_log_stride        = 20;
+            config.max_retained_block_files = 5;
+         },
+         true);
+   chain.produce_blocks(75);
+
+   BOOST_CHECK( chain.control->fetch_block_by_number(1)->block_num() == 1 );
+   BOOST_CHECK( chain.control->fetch_block_by_number(21)->block_num() == 21 );
+   BOOST_CHECK( chain.control->fetch_block_by_number(41)->block_num() == 41 );
+   BOOST_CHECK( chain.control->fetch_block_by_number(75)->block_num() == 75 );
+}
+
+void trim_blocklog_front(uint32_t version) {
+   blocklog_version_setter set_version(version); 
    tester chain;
    chain.produce_blocks(10);
    chain.produce_blocks(20);
@@ -256,6 +483,8 @@ void trim_blocklog_front() {
 
    block_log old_log(blocks_dir);
    block_log new_log(temp1.path);
+   // double check if the version has been set to the desired version
+   BOOST_CHECK(old_log.version() == version); 
    BOOST_CHECK(new_log.first_block_num() == 10);
    BOOST_CHECK(new_log.head()->block_num() == old_log.head()->block_num());
 
@@ -263,21 +492,20 @@ void trim_blocklog_front() {
    BOOST_CHECK(fc::file_size(temp1.path / "blocks.index") == old_index_size - sizeof(uint64_t) * num_blocks_trimmed);
 }
 
-BOOST_AUTO_TEST_CASE(test_trim_blocklog_front) { trim_blocklog_front(); }
+BOOST_AUTO_TEST_CASE(test_trim_blocklog_front) { 
+   trim_blocklog_front(block_log::max_supported_version); 
+}
 
 BOOST_AUTO_TEST_CASE(test_trim_blocklog_front_v1) {
-   block_log::set_version(1);
-   trim_blocklog_front();
+   trim_blocklog_front(1);
 }
 
 BOOST_AUTO_TEST_CASE(test_trim_blocklog_front_v2) {
-   block_log::set_version(2);
-   trim_blocklog_front();
+   trim_blocklog_front(2);
 }
 
 BOOST_AUTO_TEST_CASE(test_trim_blocklog_front_v3) {
-   block_log::set_version(3);
-   trim_blocklog_front();
+   trim_blocklog_front(3);
 }
 
 BOOST_AUTO_TEST_SUITE_END()
diff --git a/unittests/whitelist_blacklist_tests.cpp b/unittests/whitelist_blacklist_tests.cpp
index d43089cf958..06cebb3dbfb 100644
--- a/unittests/whitelist_blacklist_tests.cpp
+++ b/unittests/whitelist_blacklist_tests.cpp
@@ -1,7 +1,6 @@
 #include <eosio/chain/generated_transaction_object.hpp>
 #include <eosio/chain/resource_limits.hpp>
 #include <eosio/testing/tester.hpp>
-#include <eosio/testing/tester_network.hpp>
 
 #include <fc/variant_object.hpp>