Don't create more than one BlockArchiver for a block

Author: SpyCheese

validator/db/archiver.cpp

@@ -31,6 +31,11 @@ BlockArchiver::BlockArchiver(BlockHandle handle, td::actor::ActorId<ArchiveManag
 
 void BlockArchiver::start_up() {
   VLOG(VALIDATOR_DEBUG) << "started block archiver for " << handle_->id().to_str();
+  if (handle_->moved_to_archive()) {
+    VLOG(VALIDATOR_DEBUG) << "already moved";
+    finish_query();
+    return;
+  }
   if (handle_->id().is_masterchain()) {
     td::actor::send_closure(db_, &Db::get_block_state, handle_,
                             [SelfId = actor_id(this), archive = archive_](td::Result<td::Ref<ShardState>> R) {

validator/db/rootdb.cpp

@@ -388,12 +388,6 @@ void RootDb::try_get_static_file(FileHash file_hash, td::Promise<td::BufferSlice
   td::actor::send_closure(static_files_db_, &StaticFilesDb::load_file, file_hash, std::move(promise));
 }
 
-void RootDb::apply_block(BlockHandle handle, td::Promise<td::Unit> promise) {
-  td::actor::create_actor<BlockArchiver>(PSTRING() << "archiver" << handle->id().id.to_str(), std::move(handle),
-                                         archive_db_.get(), actor_id(this), std::move(promise))
-      .release();
-}
-
 void RootDb::get_block_by_lt(AccountIdPrefixFull account, LogicalTime lt, td::Promise<ConstBlockHandle> promise) {
   td::actor::send_closure(archive_db_, &ArchiveManager::get_block_by_lt, account, lt, std::move(promise));
 }
@@ -464,8 +458,26 @@ void RootDb::start_up() {
 }
 
 void RootDb::archive(BlockHandle handle, td::Promise<td::Unit> promise) {
-  td::actor::create_actor<BlockArchiver>(PSTRING() << "archiver" << handle->id().id.to_str(), std::move(handle),
-                                         archive_db_.get(), actor_id(this), std::move(promise))
+  auto [it, inserted] = archive_block_waiters_.emplace(handle->id(), std::vector<td::Promise<td::Unit>>{});
+  it->second.push_back(std::move(promise));
+  if (!inserted) {
+    VLOG(VALIDATOR_DEBUG) << "archive block " << handle->id().id.to_str() << " : already in progress";
+    return;
+  }
+  td::actor::create_actor<BlockArchiver>(
+      PSTRING() << "archiver" << handle->id().id.to_str(), handle, archive_db_.get(), actor_id(this),
+      [this, SelfId = actor_id(this), block_id = handle->id()](td::Result<td::Unit> R) {
+        td::actor::send_lambda(SelfId, [this, R = std::move(R), block_id]() {
+          auto it2 = archive_block_waiters_.find(block_id);
+          if (it2 == archive_block_waiters_.end()) {
+            return;
+          }
+          for (auto &promise : it2->second) {
+            promise.set_result(R.clone());
+          }
+          archive_block_waiters_.erase(it2);
+        });
+      })
       .release();
 }
 

validator/db/rootdb.hpp

@@ -97,7 +97,6 @@ class RootDb : public Db {
 
   void try_get_static_file(FileHash file_hash, td::Promise<td::BufferSlice> promise) override;
 
-  void apply_block(BlockHandle handle, td::Promise<td::Unit> promise) override;
   void get_block_by_lt(AccountIdPrefixFull account, LogicalTime lt, td::Promise<ConstBlockHandle> promise) override;
   void get_block_by_unix_time(AccountIdPrefixFull account, UnixTime ts, td::Promise<ConstBlockHandle> promise) override;
   void get_block_by_seqno(AccountIdPrefixFull account, BlockSeqno seqno,
@@ -157,6 +156,8 @@ class RootDb : public Db {
   td::actor::ActorOwn<StateDb> state_db_;
   td::actor::ActorOwn<StaticFilesDb> static_files_db_;
   td::actor::ActorOwn<ArchiveManager> archive_db_;
+
+  std::map<BlockIdExt, std::vector<td::Promise<td::Unit>>> archive_block_waiters_;
 };
 
 }  // namespace validator

validator/interfaces/db.h

@@ -88,7 +88,6 @@ class Db : public td::actor::Actor {
   virtual void store_block_handle(BlockHandle handle, td::Promise<td::Unit> promise) = 0;
   virtual void get_block_handle(BlockIdExt id, td::Promise<BlockHandle> promise) = 0;
 
-  virtual void apply_block(BlockHandle handle, td::Promise<td::Unit> promise) = 0;
   virtual void get_block_by_lt(AccountIdPrefixFull account, LogicalTime lt, td::Promise<ConstBlockHandle> promise) = 0;
   virtual void get_block_by_unix_time(AccountIdPrefixFull account, UnixTime ts,
                                       td::Promise<ConstBlockHandle> promise) = 0;

validator/manager-disk.cpp

@@ -856,7 +856,7 @@ void ValidatorManagerImpl::new_block(BlockHandle handle, td::Ref<ShardState> sta
                                 std::move(promise));
       }
     });
-    td::actor::send_closure(db_, &Db::apply_block, handle, std::move(P));
+    td::actor::send_closure(db_, &Db::archive, handle, std::move(P));
   }
 }
 

validator/manager.cpp

@@ -1666,7 +1666,7 @@ void ValidatorManagerImpl::new_block(BlockHandle handle, td::Ref<ShardState> sta
                                 std::move(promise));
       }
     });
-    td::actor::send_closure(db_, &Db::apply_block, handle, std::move(P));
+    td::actor::send_closure(db_, &Db::archive, handle, std::move(P));
   }
 }