Save intermediate split state parts in DB

This way, we don't need to keep full shard state at any point during
deserialization.

Author: DanShaders

validator/db/rootdb.cpp

@@ -281,6 +281,11 @@ void RootDb::get_block_state(ConstBlockHandle handle, td::Promise<td::Ref<ShardS
   }
 }
 
+void RootDb::store_block_state_part(BlockId effective_block, td::Ref<vm::Cell> cell,
+                                    td::Promise<td::Ref<vm::DataCell>> promise) {
+  td::actor::send_closure(cell_db_, &CellDb::store_cell, BlockIdExt{effective_block}, cell, std::move(promise));
+}
+
 void RootDb::get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise) {
   td::actor::send_closure(cell_db_, &CellDb::get_cell_db_reader, std::move(promise));
 }

validator/db/rootdb.hpp

@@ -63,6 +63,8 @@ class RootDb : public Db {
   void store_block_state(BlockHandle handle, td::Ref<ShardState> state,
                          td::Promise<td::Ref<ShardState>> promise) override;
   void get_block_state(ConstBlockHandle handle, td::Promise<td::Ref<ShardState>> promise) override;
+  void store_block_state_part(BlockId effective_block, td::Ref<vm::Cell> cell,
+                              td::Promise<td::Ref<vm::DataCell>> promise) override;
   void get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise) override;
 
   void store_block_handle(BlockHandle handle, td::Promise<td::Unit> promise) override;

validator/downloaders/download-state.cpp

@@ -337,7 +337,7 @@ void DownloadShardState::downloaded_split_state_header(td::BufferSlice data) {
 
   auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<td::Unit> R) {
     R.ensure();
-    td::actor::send_closure(SelfId, &DownloadShardState::written_split_state_file);
+    td::actor::send_closure(SelfId, &DownloadShardState::download_next_part_or_finish);
   });
   td::actor::send_closure(manager_, &ValidatorManager::store_persistent_state_file, block_id_, masterchain_block_id_,
                           SplitPersistentStateType{}, std::move(data), std::move(P));
@@ -347,13 +347,13 @@ namespace {
 
 void retry_part_download(td::actor::ActorId<DownloadShardState> SelfId, td::Status error) {
   LOG(WARNING) << "failed to download state part : " << error;
-  delay_action([=]() { td::actor::send_closure(SelfId, &DownloadShardState::written_split_state_file); },
+  delay_action([=]() { td::actor::send_closure(SelfId, &DownloadShardState::download_next_part_or_finish); },
                td::Timestamp::in(1.0));
 }
 
 }  // namespace
 
-void DownloadShardState::written_split_state_file() {
+void DownloadShardState::download_next_part_or_finish() {
   if (stored_parts_.size() == parts_.size()) {
     auto state_root = deserializer_->merge(stored_parts_);
     auto maybe_state = create_shard_state(block_id_, state_root);
@@ -411,7 +411,7 @@ void DownloadShardState::downloaded_state_part(td::BufferSlice data) {
 
   auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<td::Unit> R) {
     R.ensure();
-    td::actor::send_closure(SelfId, &DownloadShardState::written_split_state_file);
+    td::actor::send_closure(SelfId, &DownloadShardState::written_state_part_file);
   });
   td::actor::send_closure(manager_, &ValidatorManager::store_persistent_state_file, block_id_, masterchain_block_id_,
                           SplitAccountStateType{parts_[idx].effective_shard}, std::move(data), std::move(P));
@@ -421,6 +421,26 @@ void DownloadShardState::downloaded_state_part(td::BufferSlice data) {
                                << " out of " << parts_.size() << ")");
 }
 
+void DownloadShardState::written_state_part_file() {
+  size_t idx = stored_parts_.size() - 1;
+
+  auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<td::Ref<vm::DataCell>> R) {
+    R.ensure();
+    td::actor::send_closure(SelfId, &DownloadShardState::saved_state_part_into_celldb, R.move_as_ok());
+  });
+  td::actor::send_closure(manager_, &ValidatorManager::store_block_state_part,
+                          BlockId{block_id_.shard_full().workchain, parts_[idx].effective_shard, block_id_.seqno()},
+                          stored_parts_.back(), std::move(P));
+  LOG(INFO) << "saving to celldb state part " << idx + 1 << " out of " << parts_.size();
+  status_.set_status(PSTRING() << block_id_.id.to_str() << " : saving state part to celldb (part " << idx + 1
+                               << " out of " << parts_.size() << ")");
+}
+
+void DownloadShardState::saved_state_part_into_celldb(td::Ref<vm::DataCell> cell) {
+  stored_parts_.back() = cell;
+  download_next_part_or_finish();
+}
+
 void DownloadShardState::written_shard_state_file() {
   status_.set_status(PSTRING() << block_id_.id.to_str() << " : storing state to celldb");
   LOG(WARNING) << "written shard state file " << block_id_.to_str();

validator/downloaders/download-state.hpp

@@ -54,8 +54,10 @@ class DownloadShardState : public td::actor::Actor {
   void checked_shard_state();
 
   void downloaded_split_state_header(td::BufferSlice data);
-  void written_split_state_file();
+  void download_next_part_or_finish();
   void downloaded_state_part(td::BufferSlice data);
+  void written_state_part_file();
+  void saved_state_part_into_celldb(td::Ref<vm::DataCell> cell);
 
   void written_shard_state_file();
   void written_shard_state(td::Ref<ShardState> state);

validator/interfaces/db.h

@@ -52,6 +52,8 @@ class Db : public td::actor::Actor {
   virtual void store_block_state(BlockHandle handle, td::Ref<ShardState> state,
                                  td::Promise<td::Ref<ShardState>> promise) = 0;
   virtual void get_block_state(ConstBlockHandle handle, td::Promise<td::Ref<ShardState>> promise) = 0;
+  virtual void store_block_state_part(BlockId effective_block, td::Ref<vm::Cell> cell,
+                                      td::Promise<td::Ref<vm::DataCell>> promise) = 0;
   virtual void get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise) = 0;
 
   virtual void store_persistent_state_file(BlockIdExt block_id, BlockIdExt masterchain_block_id,

validator/interfaces/validator-manager.h

@@ -70,6 +70,8 @@ class ValidatorManager : public ValidatorManagerInterface {
   }
   virtual void set_block_state(BlockHandle handle, td::Ref<ShardState> state,
                                td::Promise<td::Ref<ShardState>> promise) = 0;
+  virtual void store_block_state_part(BlockId effective_block, td::Ref<vm::Cell> cell,
+                                      td::Promise<td::Ref<vm::DataCell>> promise) = 0;
   virtual void get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise) = 0;
   virtual void store_persistent_state_file(BlockIdExt block_id, BlockIdExt masterchain_block_id,
                                            PersistentStateType type, td::BufferSlice state,

validator/manager-disk.cpp

@@ -688,6 +688,11 @@ void ValidatorManagerImpl::set_block_state(BlockHandle handle, td::Ref<ShardStat
   td::actor::send_closure(db_, &Db::store_block_state, handle, state, std::move(promise));
 }
 
+void ValidatorManagerImpl::store_block_state_part(BlockId effective_block, td::Ref<vm::Cell> cell,
+                                                  td::Promise<td::Ref<vm::DataCell>> promise) {
+  td::actor::send_closure(db_, &Db::store_block_state_part, effective_block, cell, std::move(promise));
+}
+
 void ValidatorManagerImpl::get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise) {
   td::actor::send_closure(db_, &Db::get_cell_db_reader, std::move(promise));
 }

validator/manager-disk.hpp

@@ -150,6 +150,8 @@ class ValidatorManagerImpl : public ValidatorManager {
 
   void set_block_state(BlockHandle handle, td::Ref<ShardState> state,
                        td::Promise<td::Ref<ShardState>> promise) override;
+  void store_block_state_part(BlockId effective_block, td::Ref<vm::Cell> cell,
+                              td::Promise<td::Ref<vm::DataCell>> promise) override;
   void get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise) override;
   void store_persistent_state_file(BlockIdExt block_id, BlockIdExt masterchain_block_id, PersistentStateType type,
                                    td::BufferSlice state, td::Promise<td::Unit> promise) override;

validator/manager-hardfork.hpp

@@ -176,6 +176,11 @@ class ValidatorManagerImpl : public ValidatorManager {
                        td::Promise<td::Ref<ShardState>> promise) override {
     UNREACHABLE();
   }
+  void store_block_state_part(BlockId effective_block, td::Ref<vm::Cell> cell,
+                              td::Promise<td::Ref<vm::DataCell>> promise) override {
+    UNREACHABLE();
+  }
+
   void get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise) override;
   void store_persistent_state_file(BlockIdExt block_id, BlockIdExt masterchain_block_id, PersistentStateType type,
                                    td::BufferSlice state, td::Promise<td::Unit> promise) override {

validator/manager.cpp

@@ -1233,6 +1233,11 @@ void ValidatorManagerImpl::set_block_state(BlockHandle handle, td::Ref<ShardStat
   td::actor::send_closure(db_, &Db::store_block_state, handle, state, std::move(P));
 }
 
+void ValidatorManagerImpl::store_block_state_part(BlockId effective_block, td::Ref<vm::Cell> cell,
+                                                  td::Promise<td::Ref<vm::DataCell>> promise) {
+  td::actor::send_closure(db_, &Db::store_block_state_part, effective_block, cell, std::move(promise));
+}
+
 void ValidatorManagerImpl::get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise) {
   td::actor::send_closure(db_, &Db::get_cell_db_reader, std::move(promise));
 }