Merge pull request #9772 from EOSIO/undo_stack_initialization

Saving and restoring undo stack at shutdown and startup

libraries/chain/combined_database.cpp

@@ -232,7 +232,7 @@ namespace eosio { namespace chain {
             auto rdb        = std::shared_ptr<rocksdb::DB>{ p };
             return std::make_unique<rocks_db_type>(eosio::session::make_session(std::move(rdb), 1024));
          }() },
-         kv_undo_stack(std::make_unique<eosio::session::undo_stack<rocks_db_type>>(*kv_database)),
+         kv_undo_stack(std::make_unique<eosio::session::undo_stack<rocks_db_type>>(*kv_database, cfg.state_dir)),
          kv_snapshot_batch_threashold(cfg.persistent_storage_mbytes_batch * 1024 * 1024)  {}
 
    void combined_database::check_backing_store_setting(bool clean_startup) {
@@ -272,6 +272,20 @@ namespace eosio { namespace chain {
       }
    }
 
+   int64_t combined_database::revision() {
+      if (backing_store == backing_store_type::ROCKSDB) {
+         try {
+            try {
+                return kv_undo_stack->revision();
+            }
+            FC_LOG_AND_RETHROW()
+         }
+         CATCH_AND_EXIT_DB_FAILURE()
+      } else {
+        return db.revision();
+      }
+   }
+
    void combined_database::undo() {
       db.undo();
 

libraries/chain/controller.cpp

@@ -487,13 +487,8 @@ struct controller_impl {
 
    void startup(std::function<void()> shutdown, std::function<bool()> check_shutdown, const snapshot_reader_ptr& snapshot) {
       EOS_ASSERT( snapshot, snapshot_exception, "No snapshot reader provided" );
-# warning TODO: Chain_kv needs to fix kv_undo_stack's revision after restar
-      // Currently kv_undo_stack returns a wrong revision after restart,
-      // failing tests/terminate-scenarios-test.py.
-      // As chain_kv is being rewritten, comment out this check for now
-      // and retest after chain_kv rewriting is finished.
-      //EOS_ASSERT( db.revision() == kv_undo_stack.revision(), database_revision_mismatch_exception,
-      //            "chainbase is at revision ${a}, but chain-kv is at revision ${b}", ("a", db.revision())("b", kv_undo_stack.revision()) );
+      EOS_ASSERT( db.revision() == kv_db.revision(), database_revision_mismatch_exception,
+                  "chainbase is at revision ${a}, but chain-kv is at revision ${b}", ("a", db.revision())("b", kv_db.revision()) );
       this->shutdown = shutdown;
       ilog( "Starting initialization from snapshot, this may take a significant amount of time" );
       try {
@@ -526,13 +521,8 @@ struct controller_impl {
 
    void startup(std::function<void()> shutdown, std::function<bool()> check_shutdown, const genesis_state& genesis) {
       EOS_ASSERT( db.revision() < 1, database_exception, "This version of controller::startup only works with a fresh state database." );
-# warning TODO: Chain_kv needs to fix kv_undo_stack's revision
-      // Currently kv_undo_stack returns a wrong revision after restart,
-      // failing tests/terminate-scenarios-test.py.
-      // As chain_kv is being rewritten, comment out this check for now
-      // and retest after chain_kv rewriting is finished.
-      //EOS_ASSERT( db.revision() == kv_undo_stack.revision(), database_revision_mismatch_exception,
-      //            "chainbase is at revision ${a}, but chain-kv is at revision ${b}", ("a", db.revision())("b", kv_undo_stack.revision()) );
+      EOS_ASSERT( db.revision() == kv_db.revision(), database_revision_mismatch_exception,
+                  "chainbase is at revision ${a}, but chain-kv is at revision ${b}", ("a", db.revision())("b", kv_db.revision()) );
       const auto& genesis_chain_id = genesis.compute_chain_id();
       EOS_ASSERT( genesis_chain_id == chain_id, chain_id_type_exception,
                   "genesis state provided to startup corresponds to a chain ID (${genesis_chain_id}) that does not match the chain ID that controller was constructed with (${controller_chain_id})",
@@ -566,9 +556,8 @@ struct controller_impl {
 
    void startup(std::function<void()> shutdown, std::function<bool()> check_shutdown) {
       EOS_ASSERT( db.revision() >= 1, database_exception, "This version of controller::startup does not work with a fresh state database." );
-# warning TODO: Chain_kv needs to fix kv_undo_stack's revision
-      //EOS_ASSERT( db.revision() == kv_undo_stack.revision(), database_revision_mismatch_exception,
-      //            "chainbase is at revision ${a}, but chain-kv is at revision ${b}", ("a", db.revision())("b", kv_undo_stack.revision()) );
+      EOS_ASSERT( db.revision() == kv_db.revision(), database_revision_mismatch_exception,
+                  "chainbase is at revision ${a}, but chain-kv is at revision ${b}", ("a", db.revision())("b", kv_db.revision()) );
       EOS_ASSERT( fork_db.head(), fork_database_exception, "No existing fork database despite existing chain state. Replay required." );
 
       this->shutdown = shutdown;

libraries/chain/include/eosio/chain/combined_database.hpp

@@ -105,6 +105,8 @@ namespace eosio { namespace chain {
 
       void set_revision(uint64_t revision);
 
+      int64_t revision();
+
       void undo();
 
       void commit(int64_t revision);

libraries/chain_kv/include/b1/session/session.hpp

@@ -259,7 +259,7 @@ class session {
    It& first_not_deleted_in_iterator_cache_(It& it, const It& end) const;
 
  private:
-   parent_variant_type m_parent{ nullptr };
+   parent_variant_type m_parent{ static_cast<Parent*>(nullptr) };
    cache_type          m_cache;
 };
 

libraries/chain_kv/include/b1/session/shared_bytes.hpp

@@ -11,6 +11,8 @@
 #include <boost/endian/detail/intrinsic.hpp>
 
 #include <fc/crypto/base64.hpp>
+#include <fc/io/datastream.hpp>
+#include <fc/io/raw.hpp>
 
 #include <eosio/chain/exceptions.hpp>
 
@@ -365,6 +367,23 @@ inline std::ostream& operator<<(std::ostream& os, const shared_bytes& bytes) {
    return os;
 }
 
+template <typename Stream>
+inline Stream& operator<<(Stream& ds, const shared_bytes& b) {
+   fc::raw::pack( ds, b.size() );
+   ds.write(b.data(), b.size());
+   return ds;
+}
+
+template <typename Stream>
+inline Stream& operator>>(Stream& ds, shared_bytes& b) {
+   std::size_t sz;
+   fc::raw::unpack( ds, sz );
+   shared_bytes tmp = {sz};
+   ds.read(tmp.data(), tmp.size());
+   b = tmp;
+   return ds;
+}
+
 inline shared_bytes shared_bytes::from_hex_string(const std::string& str) {
    if (str.empty()) {
       return shared_bytes{};

libraries/chain_kv/include/b1/session/undo_stack.hpp

@@ -2,11 +2,20 @@
 
 #include <queue>
 
+#include <fc/filesystem.hpp>
+#include <fc/io/fstream.hpp>
+#include <fc/io/datastream.hpp>
+#include <fc/io/raw.hpp>
+#include <fstream>
 #include <b1/session/session.hpp>
 #include <b1/session/session_variant.hpp>
 #include <eosio/chain/exceptions.hpp>
 
 namespace eosio::session {
+   constexpr uint32_t undo_stack_magic_number = 0x30510ABC;
+   constexpr uint32_t undo_stack_min_supported_version = 1;
+   constexpr uint32_t undo_stack_max_supported_version = 1;
+   constexpr auto undo_stack_filename = "undo_stack.dat";
 
 /// \brief Represents a container of pending sessions to be committed.
 template <typename Session>
@@ -19,7 +28,7 @@ class undo_stack {
 
    /// \brief Constructor.
    /// \param head The session that the changes are merged into when commit is called.
-   undo_stack(Session& head);
+   undo_stack(Session& head, const fc::path& datadir = {});
    undo_stack(const undo_stack&) = delete;
    undo_stack(undo_stack&&)      = default;
    ~undo_stack();
@@ -66,19 +75,25 @@ class undo_stack {
    /// \remarks This is the next session to be committed.
    const_variant_type bottom() const;
 
+   void open();
+   void close();
+
  private:
    int64_t                  m_revision{ 0 };
    Session*                 m_head;
    std::deque<session_type> m_sessions; // Need a deque so pointers don't become invalidated.  The session holds a
                                         // pointer to the parent internally.
+   fc::path                 m_datadir;
 };
 
 template <typename Session>
-undo_stack<Session>::undo_stack(Session& head) : m_head{ &head } {}
+undo_stack<Session>::undo_stack(Session& head, const fc::path& datadir) : m_head{ &head }, m_datadir{ datadir } {
+   open();
+}
 
 template <typename Session>
 undo_stack<Session>::~undo_stack() {
-   for (auto& session : m_sessions) { session.undo(); }
+   close();
 }
 
 template <typename Session>
@@ -197,4 +212,113 @@ typename undo_stack<Session>::const_variant_type undo_stack<Session>::bottom() c
    return { *m_head, nullptr };
 }
 
+template <typename Session>
+void undo_stack<Session>::open() {
+   if (m_datadir.empty())
+      return;
+
+   if (!fc::is_directory(m_datadir))
+      fc::create_directories(m_datadir);
+
+   auto undo_stack_dat = m_datadir / undo_stack_filename;
+   if( fc::exists( undo_stack_dat ) ) {
+      try {
+         std::string content;
+         fc::read_file_contents( undo_stack_dat, content );
+
+         fc::datastream<const char*> ds( content.data(), content.size() );
+
+         // validate totem
+         uint32_t totem = 0;
+         fc::raw::unpack( ds, totem );
+         EOS_ASSERT( totem == undo_stack_magic_number, eosio::chain::chain_exception,
+                     "Undo stack data file '${filename}' has unexpected magic number: ${actual_totem}. Expected ${expected_totem}",
+                     ("filename", undo_stack_dat.generic_string())
+                     ("actual_totem", totem)
+                     ("expected_totem", undo_stack_magic_number)
+         );
+
+         // validate version
+         uint32_t version = 0;
+         fc::raw::unpack( ds, version );
+         EOS_ASSERT( version >= undo_stack_min_supported_version && version <= undo_stack_max_supported_version,
+                    eosio::chain::chain_exception,
+                    "Unsupported version of Undo stack data file '${filename}'. "
+                    "Undo stack data version is ${version} while code supports version(s) [${min},${max}]",
+                    ("filename", undo_stack_dat.generic_string())
+                    ("version", version)
+                    ("min", undo_stack_min_supported_version)
+                    ("max", undo_stack_max_supported_version)
+         );
+
+         int64_t rev; fc::raw::unpack( ds, rev );
+
+         size_t num_sessions; fc::raw::unpack( ds, num_sessions );
+         for( size_t i = 0; i < num_sessions; ++i ) {
+            push();
+            auto& session = m_sessions.back();
+
+            size_t num_updated_keys; fc::raw::unpack( ds, num_updated_keys );
+            for( size_t j = 0; j < num_updated_keys; ++j ) {
+               shared_bytes key; ds >> key;
+               shared_bytes value; ds >> value;
+               session.write(key, value);
+            }
+
+            size_t num_deleted_keys; fc::raw::unpack( ds, num_deleted_keys );
+            for( size_t j = 0; j < num_deleted_keys; ++j ) {
+               shared_bytes key; ds >> key;
+               session.erase(key);
+            }
+         }
+         m_revision = rev; // restore head revision
+      } FC_CAPTURE_AND_RETHROW( (undo_stack_dat) )
+
+      fc::remove( undo_stack_dat );
+   }
+}
+
+template <typename Session>
+void undo_stack<Session>::close() {
+   if (m_datadir.empty())
+      return;
+
+   auto undo_stack_dat = m_datadir / undo_stack_filename;
+
+   std::ofstream out( undo_stack_dat.generic_string().c_str(), std::ios::out | std::ios::binary | std::ofstream::trunc );
+   fc::raw::pack( out, undo_stack_magic_number );
+   fc::raw::pack( out, undo_stack_max_supported_version );
+
+   fc::raw::pack( out, revision() );
+   fc::raw::pack( out, size() ); // number of sessions
+
+   while ( !m_sessions.empty() ) {
+      auto& session = m_sessions.front();
+      auto updated_keys = session.updated_keys();
+      fc::raw::pack( out, updated_keys.size() ); // number of updated keys
+
+      for (const auto& key: updated_keys) {
+         auto value = session.read(key);
+
+         if ( value ) {
+            out << key;
+            out << *value;
+         } else {
+            fc::remove( undo_stack_dat ); // May not be used by next startup
+            elog( "Did not find value for ${k}", ("k", key.data() ) );
+            return; // Do not assert as we are during shutdown
+         }
+      }
+
+      auto deleted_keys = session.deleted_keys();
+      fc::raw::pack( out, deleted_keys.size() ); // number of deleted keys
+
+      for (const auto& key: deleted_keys) {
+         fc::raw::pack( out, key );
+      }
+
+      session.detach();
+      m_sessions.pop_front();
+   }
+}
 } // namespace eosio::session