Optimize importing out queues

validator/impl/collator-impl.h

@@ -233,7 +233,8 @@ class Collator final : public td::actor::Actor {
   void after_get_aux_shard_state(ton::BlockIdExt blkid, td::Result<Ref<ShardState>> res);
   bool fix_one_processed_upto(block::MsgProcessedUpto& proc, const ton::ShardIdFull& owner);
   bool fix_processed_upto(block::MsgProcessedUptoCollection& upto);
-  void got_neighbor_msg_queue(unsigned i, td::Result<Ref<OutMsgQueueProof>> R);
+  void got_neighbor_msg_queues(td::Result<std::map<BlockIdExt, Ref<OutMsgQueueProof>>> R);
+  void got_neighbor_msg_queue(unsigned i, Ref<OutMsgQueueProof> res);
   bool adjust_shard_config();
   bool store_shard_fees(ShardIdFull shard, const block::CurrencyCollection& fees,
                         const block::CurrencyCollection& created);

validator/impl/collator.cpp

@@ -632,27 +632,45 @@ bool Collator::request_neighbor_msg_queues() {
     }
     neighbors_.emplace_back(*shard_ptr);
   }
+  std::vector<BlockIdExt> top_blocks;
   unsigned i = 0;
   for (block::McShardDescr& descr : neighbors_) {
     LOG(DEBUG) << "neighbor #" << i << " : " << descr.blk_.to_str();
-    ++pending;
-    send_closure_later(manager, &ValidatorManager::wait_out_msg_queue_proof, descr.blk_, shard_, priority(), timeout,
-                       [self = get_self(), i](td::Result<Ref<OutMsgQueueProof>> res) {
-                         LOG(DEBUG) << "got msg queue for neighbor #" << i;
-                         send_closure_later(std::move(self), &Collator::got_neighbor_msg_queue, i, std::move(res));
-                       });
+    top_blocks.push_back(descr.blk_);
     ++i;
   }
+  ++pending;
+  td::actor::send_closure_later(
+      manager, &ValidatorManager::wait_neighbor_msg_queue_proofs, shard_, std::move(top_blocks), timeout,
+      [self = get_self()](td::Result<std::map<BlockIdExt, Ref<OutMsgQueueProof>>> res) {
+        td::actor::send_closure_later(std::move(self), &Collator::got_neighbor_msg_queues, std::move(res));
+      });
   return true;
 }
 
-void Collator::got_neighbor_msg_queue(unsigned i, td::Result<Ref<OutMsgQueueProof>> R) {
+void Collator::got_neighbor_msg_queues(td::Result<std::map<BlockIdExt, Ref<OutMsgQueueProof>>> R) {
   --pending;
   if (R.is_error()) {
-    fatal_error(R.move_as_error());
+    fatal_error(R.move_as_error_prefix("failed to get neighbor msg queues: "));
     return;
   }
+  LOG(INFO) << "neighbor output queues fetched";
   auto res = R.move_as_ok();
+  unsigned i = 0;
+  for (block::McShardDescr& descr : neighbors_) {
+    LOG(DEBUG) << "neighbor #" << i << " : " << descr.blk_.to_str();
+    auto it = res.find(descr.blk_);
+    if (it == res.end()) {
+      fatal_error(PSTRING() << "no msg queue from neighbor #" << i);
+      return;
+    }
+    got_neighbor_msg_queue(i, it->second);
+    ++i;
+  }
+  check_pending();
+}
+
+void Collator::got_neighbor_msg_queue(unsigned i, Ref<OutMsgQueueProof> res) {
   BlockIdExt block_id = neighbors_.at(i).blk_;
   if (res->block_state_proof_.not_null() && !block_id.is_masterchain()) {
     block_state_proofs_.emplace(block_id.root_hash, res->block_state_proof_);
@@ -732,10 +750,6 @@ void Collator::got_neighbor_msg_queue(unsigned i, td::Result<Ref<OutMsgQueueProo
       }
     }
   } while (false);
-  if (!pending) {
-    LOG(INFO) << "all neighbor output queues fetched";
-  }
-  check_pending();
 }
 
 bool Collator::unpack_merge_last_state() {

validator/impl/out-msg-queue-proof.cpp

@@ -46,9 +46,9 @@ static td::Status check_no_prunned(const vm::CellSlice& cs) {
   return td::Status::OK();
 }
 
-static td::Result<td::int32> process_queue(BlockIdExt block_id, ShardIdFull dst_shard,
-                                           block::ImportedMsgQueueLimits limits,
-                                           const block::gen::OutMsgQueueInfo::Record& qinfo) {
+static td::Result<std::vector<td::int32>> process_queue(
+    ShardIdFull dst_shard, std::vector<std::pair<BlockIdExt, block::gen::OutMsgQueueInfo::Record>> blocks,
+    block::ImportedMsgQueueLimits limits) {
   td::uint64 estimated_proof_size = 0;
 
   td::HashSet<vm::Cell::Hash> visited;
@@ -66,14 +66,18 @@ static td::Result<td::int32> process_queue(BlockIdExt block_id, ShardIdFull dst_
       dfs(cs.prefetch_ref(i));
     }
   };
-  TRY_STATUS_PREFIX(check_no_prunned(*qinfo.proc_info), "invalid proc_info proof: ")
-  TRY_STATUS_PREFIX(check_no_prunned(*qinfo.ihr_pending), "invalid ihr_pending proof: ")
-  dfs_cs(*qinfo.proc_info);
-  dfs_cs(*qinfo.ihr_pending);
+  std::vector<block::OutputQueueMerger::Neighbor> neighbors;
+  for (auto& b : blocks) {
+    TRY_STATUS_PREFIX(check_no_prunned(*b.second.proc_info), "invalid proc_info proof: ")
+    TRY_STATUS_PREFIX(check_no_prunned(*b.second.ihr_pending), "invalid ihr_pending proof: ")
+    dfs_cs(*b.second.proc_info);
+    dfs_cs(*b.second.ihr_pending);
+    neighbors.emplace_back(b.first, b.second.out_queue->prefetch_ref());
+  }
 
-  block::OutputQueueMerger queue_merger{
-      dst_shard, {block::OutputQueueMerger::Neighbor{block_id, qinfo.out_queue->prefetch_ref()}}};
-  td::int32 msg_count = 0;
+  block::OutputQueueMerger queue_merger{dst_shard, std::move(neighbors)};
+  std::vector<td::int32> msg_count(blocks.size());
+  td::int32 msg_count_total = 0;
   bool limit_reached = false;
 
   while (!queue_merger.is_eof()) {
@@ -87,7 +91,8 @@ static td::Result<td::int32> process_queue(BlockIdExt block_id, ShardIdFull dst_
     if (limit_reached) {
       break;
     }
-    ++msg_count;
+    ++msg_count[kv->source];
+    ++msg_count_total;
 
     // TODO: Get processed_upto from destination shard (in request?)
     /*
@@ -114,234 +119,410 @@ static td::Result<td::int32> process_queue(BlockIdExt block_id, ShardIdFull dst_
 
     dfs_cs(*kv->msg);
     TRY_STATUS_PREFIX(check_no_prunned(*kv->msg), "invalid message proof: ")
-    if (estimated_proof_size >= limits.max_bytes || msg_count >= (long long)limits.max_msgs) {
+    if (estimated_proof_size >= limits.max_bytes || msg_count_total >= (long long)limits.max_msgs) {
       limit_reached = true;
     }
   }
-  return limit_reached ? msg_count : -1;
+  if (!limit_reached) {
+    std::fill(msg_count.begin(), msg_count.end(), -1);
+  }
+  return msg_count;
 }
 
 td::Result<tl_object_ptr<ton_api::tonNode_outMsgQueueProof>> OutMsgQueueProof::build(
-    BlockIdExt block_id, ShardIdFull dst_shard,
-    block::ImportedMsgQueueLimits limits, Ref<vm::Cell> state_root, Ref<vm::Cell> block_root) {
+    ShardIdFull dst_shard, std::vector<OneBlock> blocks, block::ImportedMsgQueueLimits limits) {
   if (!dst_shard.is_valid_ext()) {
     return td::Status::Error("invalid shard");
   }
-
-  vm::MerkleProofBuilder mpb{std::move(state_root)};
-  TRY_RESULT(state, ShardStateQ::fetch(block_id, {}, mpb.root()));
-  TRY_RESULT(outq_descr, state->message_queue());
-  block::gen::OutMsgQueueInfo::Record qinfo;
-  if (!tlb::unpack_cell(outq_descr->root_cell(), qinfo)) {
-    return td::Status::Error("invalid message queue");
+  if (blocks.empty()) {
+    return create_tl_object<ton_api::tonNode_outMsgQueueProof>(td::BufferSlice{}, td::BufferSlice{},
+                                                               std::vector<td::int32>{});
   }
-  TRY_RESULT(cnt, process_queue(block_id, dst_shard, limits, qinfo));
 
-  TRY_RESULT(queue_proof, mpb.extract_proof_boc());
-  td::BufferSlice block_state_proof;
-  if (block_id.seqno() != 0) {
-    TRY_RESULT(proof, create_block_state_proof(std::move(block_root)));
-    TRY_RESULT_ASSIGN(block_state_proof, vm::std_boc_serialize(std::move(proof), 31));
+  std::vector<td::Ref<vm::Cell>> block_state_proofs;
+  for (auto& block : blocks) {
+    if (block.id.seqno() != 0) {
+      if (block.block_root.is_null()) {
+        return td::Status::Error("block is null");
+      }
+      TRY_RESULT(proof, create_block_state_proof(block.block_root));
+      block_state_proofs.push_back(std::move(proof));
+    }
+    if (!block::ShardConfig::is_neighbor(dst_shard, block.id.shard_full())) {
+      return td::Status::Error("shards are not neighbors");
+    }
   }
+  TRY_RESULT(block_state_proof, vm::std_boc_serialize_multi(block_state_proofs));
+
+  vm::Dictionary states_dict_pure{32};
+  for (size_t i = 0; i < blocks.size(); ++i) {
+    if (blocks[i].state_root.is_null()) {
+      return td::Status::Error("state is null");
+    }
+    states_dict_pure.set_ref(td::BitArray<32>{(long long)i}, blocks[i].state_root);
+  }
+
+  vm::MerkleProofBuilder mpb{states_dict_pure.get_root_cell()};
+  vm::Dictionary states_dict{mpb.root(), 32};
+  std::vector<std::pair<BlockIdExt, block::gen::OutMsgQueueInfo::Record>> data(blocks.size());
+  for (size_t i = 0; i < blocks.size(); ++i) {
+    data[i].first = blocks[i].id;
+    TRY_RESULT(state, ShardStateQ::fetch(blocks[i].id, {}, states_dict.lookup_ref(td::BitArray<32>{(long long)i})));
+    TRY_RESULT(outq_descr, state->message_queue());
+    block::gen::OutMsgQueueInfo::Record qinfo;
+    if (!tlb::unpack_cell(outq_descr->root_cell(), data[i].second)) {
+      return td::Status::Error("invalid message queue");
+    }
+  }
+  TRY_RESULT(msg_count, process_queue(dst_shard, std::move(data), limits));
+
+  TRY_RESULT(proof, mpb.extract_proof());
+  vm::Dictionary states_dict_proof{vm::CellSlice{vm::NoVm(), proof}.prefetch_ref(), 32};
+  std::vector<td::Ref<vm::Cell>> state_proofs;
+  for (size_t i = 0; i < blocks.size(); ++i) {
+    td::Ref<vm::Cell> proof_raw = states_dict_proof.lookup_ref(td::BitArray<32>{(long long)i});
+    CHECK(proof_raw.not_null());
+    state_proofs.push_back(vm::CellBuilder::create_merkle_proof(proof_raw));
+  }
+  TRY_RESULT(queue_proof, vm::std_boc_serialize_multi(state_proofs));
   return create_tl_object<ton_api::tonNode_outMsgQueueProof>(std::move(queue_proof), std::move(block_state_proof),
-                                                             cnt);
+                                                             std::move(msg_count));
 }
 
-td::Result<td::Ref<OutMsgQueueProof>> OutMsgQueueProof::fetch(BlockIdExt block_id, ShardIdFull dst_shard,
-                                                              block::ImportedMsgQueueLimits limits,
-                                                              const ton_api::tonNode_outMsgQueueProof& f) {
+td::Result<std::vector<td::Ref<OutMsgQueueProof>>> OutMsgQueueProof::fetch(ShardIdFull dst_shard,
+                                                                           std::vector<BlockIdExt> blocks,
+                                                                           block::ImportedMsgQueueLimits limits,
+                                                                           const ton_api::tonNode_outMsgQueueProof& f) {
   try {
-    Ref<vm::Cell> block_state_proof;
-    td::Bits256 state_root_hash;
-    if (block_id.seqno() == 0) {
-      if (!f.block_state_proof_.empty()) {
-        return td::Status::Error("expected empty block state proof");
+    std::vector<td::Ref<OutMsgQueueProof>> res;
+    TRY_RESULT(queue_proofs, vm::std_boc_deserialize_multi(f.queue_proofs_, (int)blocks.size()));
+    TRY_RESULT(block_state_proofs, vm::std_boc_deserialize_multi(f.block_state_proofs_, (int)blocks.size()));
+    if (queue_proofs.size() != blocks.size()) {
+      return td::Status::Error("invalid size of queue_proofs");
+    }
+    if (f.msg_counts_.size() != blocks.size()) {
+      return td::Status::Error("invalid size of msg_counts");
+    }
+    size_t j = 0;
+    std::vector<std::pair<BlockIdExt, block::gen::OutMsgQueueInfo::Record>> data(blocks.size());
+    for (size_t i = 0; i < blocks.size(); ++i) {
+      td::Bits256 state_root_hash;
+      Ref<vm::Cell> block_state_proof = {};
+      if (blocks[i].seqno() == 0) {
+        state_root_hash = blocks[i].root_hash;
+      } else {
+        if (j == block_state_proofs.size()) {
+          return td::Status::Error("invalid size of block_state_proofs");
+        }
+        block_state_proof = block_state_proofs[j++];
+        TRY_RESULT_ASSIGN(state_root_hash, unpack_block_state_proof(blocks[i], block_state_proof));
       }
-      state_root_hash = block_id.root_hash;
-    } else {
-      TRY_RESULT_ASSIGN(block_state_proof, vm::std_boc_deserialize(f.block_state_proof_.as_slice()));
-      TRY_RESULT_ASSIGN(state_root_hash, unpack_block_state_proof(block_id, block_state_proof));
-    }
+      auto state_root = vm::MerkleProof::virtualize(queue_proofs[i], 1);
+      if (state_root->get_hash().as_slice() != state_root_hash.as_slice()) {
+        return td::Status::Error("state root hash mismatch");
+      }
+      res.emplace_back(true, blocks[i], state_root, block_state_proof, f.msg_counts_[i]);
 
-    TRY_RESULT(queue_proof, vm::std_boc_deserialize(f.queue_proof_.as_slice()));
-    auto virtual_root = vm::MerkleProof::virtualize(queue_proof, 1);
-    if (virtual_root.is_null()) {
-      return td::Status::Error("invalid queue proof");
+      data[i].first = blocks[i];
+      TRY_RESULT(state, ShardStateQ::fetch(blocks[i], {}, state_root));
+      TRY_RESULT(outq_descr, state->message_queue());
+      block::gen::OutMsgQueueInfo::Record qinfo;
+      if (!tlb::unpack_cell(outq_descr->root_cell(), data[i].second)) {
+        return td::Status::Error("invalid message queue");
+      }
     }
-    if (virtual_root->get_hash().as_slice() != state_root_hash.as_slice()) {
-      return td::Status::Error("state root hash mismatch");
+    if (j != block_state_proofs.size()) {
+      return td::Status::Error("invalid size of block_state_proofs");
     }
-
-    // Validate proof
-    TRY_RESULT_PREFIX(state, ShardStateQ::fetch(block_id, {}, virtual_root), "invalid proof: ");
-    TRY_RESULT_PREFIX(outq_descr, state->message_queue(), "invalid proof: ");
-
-    block::gen::OutMsgQueueInfo::Record qinfo;
-    if (!tlb::unpack_cell(outq_descr->root_cell(), qinfo)) {
-      return td::Status::Error("invalid proof: invalid message queue");
+    TRY_RESULT(msg_count, process_queue(dst_shard, std::move(data), limits));
+    if (msg_count != f.msg_counts_) {
+      return td::Status::Error("incorrect msg_count");
     }
-    TRY_STATUS_PREFIX(check_no_prunned(qinfo.proc_info->prefetch_ref(0)), "invalid proc_info: ")
-    TRY_STATUS_PREFIX(check_no_prunned(qinfo.ihr_pending->prefetch_ref(0)), "invalid ihr_pending: ")
-    TRY_RESULT(cnt, process_queue(block_id, dst_shard, limits, qinfo));
-    if (cnt != f.msg_count_) {
-      return td::Status::Error(PSTRING() << "invalid msg_count: expected=" << f.msg_count_ << ", found=" << cnt);
-    }
-    return Ref<OutMsgQueueProof>(true, std::move(virtual_root), std::move(block_state_proof), cnt);
+    return res;
   } catch (vm::VmVirtError& err) {
     return td::Status::Error(PSTRING() << "invalid proof: " << err.get_msg());
   }
 }
 
-void WaitOutMsgQueueProof::alarm() {
-  abort_query(td::Status::Error(ErrorCode::timeout, "timeout"));
-}
-
-void WaitOutMsgQueueProof::abort_query(td::Status reason) {
-  if (promise_) {
-    LOG(DEBUG) << "aborting wait msg queue query for " << block_id_.to_str() << " priority=" << priority_ << ": "
-               << reason;
-    promise_.set_error(
-        reason.move_as_error_prefix(PSTRING() << "failed to get msg queue for " << block_id_.to_str() << ": "));
+void OutMsgQueueImporter::new_masterchain_block_notification(td::Ref<MasterchainState> state,
+                                                             std::set<ShardIdFull> collating_shards) {
+  last_masterchain_state_ = state;
+  if (collating_shards.empty() || state->get_unix_time() < (td::uint32)td::Clocks::system() - 20) {
+    return;
   }
-  stop();
-}
-
-void WaitOutMsgQueueProof::finish_query(Ref<OutMsgQueueProof> result) {
-  promise_.set_result(std::move(result));
-  stop();
-}
-
-void WaitOutMsgQueueProof::start_up() {
-  alarm_timestamp() = timeout_;
-  if (local_) {
-    run_local();
-  } else {
-    run_net();
-  }
-}
-
-void WaitOutMsgQueueProof::run_local() {
-  ++pending;
-  td::actor::send_closure(manager_, &ValidatorManager::wait_block_state_short, block_id_, priority_, timeout_,
-                          [SelfId = actor_id(this)](td::Result<Ref<ShardState>> R) {
-                            if (R.is_error()) {
-                              td::actor::send_closure(SelfId, &WaitOutMsgQueueProof::abort_query,
-                                                      R.move_as_error_prefix("failed to get shard state: "));
-                            } else {
-                              td::actor::send_closure(SelfId, &WaitOutMsgQueueProof::got_state_root,
-                                                      R.move_as_ok()->root_cell());
-                            }
-                          });
-  if (block_id_.seqno() != 0) {
-    ++pending;
-    td::actor::send_closure(manager_, &ValidatorManager::wait_block_data_short, block_id_, priority_, timeout_,
-                            [SelfId = actor_id(this)](td::Result<Ref<BlockData>> R) {
-                              if (R.is_error()) {
-                                td::actor::send_closure(SelfId, &WaitOutMsgQueueProof::abort_query,
-                                                        R.move_as_error_prefix("failed to get block data: "));
-                              } else {
-                                td::actor::send_closure(SelfId, &WaitOutMsgQueueProof::got_block_root,
-                                                        R.move_as_ok()->root_cell());
-                              }
-                            });
-  }
-}
-
-void WaitOutMsgQueueProof::got_state_root(Ref<vm::Cell> root) {
-  state_root_ = std::move(root);
-  if (--pending == 0) {
-    run_local_cont();
-  }
-}
-
-void WaitOutMsgQueueProof::got_block_root(Ref<vm::Cell> root) {
-  block_root_ = std::move(root);
-  if (--pending == 0) {
-    run_local_cont();
-  }
-}
-
-void WaitOutMsgQueueProof::run_local_cont() {
-  Ref<vm::Cell> block_state_proof;
-  if (block_id_.seqno() != 0) {
-    auto R = create_block_state_proof(std::move(block_root_));
-    if (R.is_error()) {
-      abort_query(R.move_as_error_prefix("failed to create block state proof"));
+  auto can_collate_shard = [&](const ShardIdFull& shard) -> bool {
+    return std::any_of(collating_shards.begin(), collating_shards.end(),
+                       [&](ShardIdFull our_shard) { return shard_intersects(shard, our_shard); });
+  };
+  auto shards = state->get_shards();
+  auto process_dst_shard = [&](const ShardIdFull& dst_shard) {
+    if (!can_collate_shard(dst_shard)) {
       return;
     }
-    block_state_proof = R.move_as_ok();
+    std::vector<BlockIdExt> top_blocks;
+    for (const auto& shard : shards) {
+      if (block::ShardConfig::is_neighbor(dst_shard, shard->shard())) {
+        top_blocks.push_back(shard->top_block_id());
+      }
+    }
+    get_neighbor_msg_queue_proofs(dst_shard, std::move(top_blocks), td::Timestamp::in(15.0),
+                                  [](td::Result<std::map<BlockIdExt, td::Ref<OutMsgQueueProof>>>) {});
+  };
+  for (const auto& shard : shards) {
+    if (shard->before_merge()) {
+      if (is_left_child(shard->shard())) {
+        process_dst_shard(shard_parent(shard->shard()));
+      }
+    } else if (shard->before_split()) {
+      process_dst_shard(shard_child(shard->shard(), true));
+      process_dst_shard(shard_child(shard->shard(), false));
+    } else {
+      process_dst_shard(shard->shard());
+    }
   }
-  finish_query(td::Ref<OutMsgQueueProof>(true, std::move(state_root_), std::move(block_state_proof)));
 }
 
-void WaitOutMsgQueueProof::run_net() {
-  auto P = td::PromiseCreator::lambda([SelfId = actor_id(this), block_id = block_id_,
-                                       retry_after = td::Timestamp::in(0.5)](td::Result<Ref<OutMsgQueueProof>> R) {
-    if (R.is_error()) {
-      LOG(DEBUG) << "failed to get msg queue for " << block_id.to_str() << " from net: " << R.move_as_error();
-      delay_action([SelfId]() mutable { td::actor::send_closure(SelfId, &WaitOutMsgQueueProof::run_net); },
-                   retry_after);
+void OutMsgQueueImporter::get_neighbor_msg_queue_proofs(
+    ShardIdFull dst_shard, std::vector<BlockIdExt> blocks, td::Timestamp timeout,
+    td::Promise<std::map<BlockIdExt, td::Ref<OutMsgQueueProof>>> promise) {
+  std::sort(blocks.begin(), blocks.end());
+  auto entry = cache_[{dst_shard, blocks}];
+  if (entry) {
+    if (entry->done) {
+      promise.set_result(entry->result);
+      alarm_timestamp().relax(entry->timeout = td::Timestamp::in(CACHE_TTL));
     } else {
-      td::actor::send_closure(SelfId, &WaitOutMsgQueueProof::finish_query, R.move_as_ok());
+      entry->timeout = std::max(entry->timeout, timeout);
+      entry->promises.emplace_back(std::move(promise), timeout);
+      alarm_timestamp().relax(timeout);
     }
-  });
+    return;
+  }
 
-  td::actor::send_closure(manager_, &ValidatorManager::send_get_out_msg_queue_proof_request, block_id_, dst_shard_,
-                          limits_, priority_, std::move(P));
+  LOG(DEBUG) << "Importing neighbor msg queues for shard " << dst_shard.to_str() << ", " << blocks.size() << " blocks";
+
+  cache_[{dst_shard, blocks}] = entry = std::make_shared<CacheEntry>();
+  entry->dst_shard = dst_shard;
+  entry->blocks = blocks;
+  entry->promises.emplace_back(std::move(promise), timeout);
+  alarm_timestamp().relax(entry->timeout = timeout);
+
+  std::map<ShardIdFull, std::vector<BlockIdExt>> new_queries;
+  for (const BlockIdExt& block : blocks) {
+    if (opts_->need_monitor(block.shard_full(), last_masterchain_state_)) {
+      ++entry->pending;
+      get_proof_local(entry, block);
+    } else {
+      ShardIdFull prefix = block.shard_full();
+      int min_split = last_masterchain_state_->monitor_min_split_depth(prefix.workchain);
+      if (prefix.pfx_len() > min_split) {
+        prefix = shard_prefix(prefix, min_split);
+      }
+      new_queries[prefix].push_back(block);
+    }
+  };
+  auto limits = last_masterchain_state_->get_imported_msg_queue_limits(dst_shard.workchain);
+  for (auto& p : new_queries) {
+    ++entry->pending;
+    get_proof_import(entry, std::move(p.second), limits);
+  }
+  if (entry->pending == 0) {
+    finish_query(entry);
+  }
+}
+
+void OutMsgQueueImporter::get_proof_local(std::shared_ptr<CacheEntry> entry, BlockIdExt block) {
+  if (!check_timeout(entry)) {
+    return;
+  }
+  td::actor::send_closure(
+      manager_, &ValidatorManager::wait_block_state_short, block, 0, entry->timeout,
+      [=, SelfId = actor_id(this), manager = manager_, timeout = entry->timeout,
+       retry_after = td::Timestamp::in(0.5)](td::Result<Ref<ShardState>> R) mutable {
+        if (R.is_error()) {
+          LOG(DEBUG) << "Failed to get block state for " << block.to_str() << ": " << R.move_as_error();
+          delay_action([=]() { td::actor::send_closure(SelfId, &OutMsgQueueImporter::get_proof_local, entry, block); },
+                       retry_after);
+          return;
+        }
+        auto state = R.move_as_ok();
+        if (block.seqno() == 0) {
+          std::vector<td::Ref<OutMsgQueueProof>> proof = {
+              td::Ref<OutMsgQueueProof>(true, block, state->root_cell(), td::Ref<vm::Cell>{})};
+          td::actor::send_closure(SelfId, &OutMsgQueueImporter::got_proof, entry, std::move(proof));
+          return;
+        }
+        td::actor::send_closure(
+            manager, &ValidatorManager::wait_block_data_short, block, 0, timeout,
+            [=](td::Result<Ref<BlockData>> R) mutable {
+              if (R.is_error()) {
+                LOG(DEBUG) << "Failed to get block data for " << block.to_str() << ": " << R.move_as_error();
+                delay_action(
+                    [=]() { td::actor::send_closure(SelfId, &OutMsgQueueImporter::get_proof_local, entry, block); },
+                    retry_after);
+                return;
+              }
+              Ref<vm::Cell> block_state_proof = create_block_state_proof(R.ok()->root_cell()).move_as_ok();
+              std::vector<td::Ref<OutMsgQueueProof>> proof = {
+                  td::Ref<OutMsgQueueProof>(true, block, state->root_cell(), std::move(block_state_proof))};
+              td::actor::send_closure(SelfId, &OutMsgQueueImporter::got_proof, entry, std::move(proof));
+            });
+      });
+}
+
+void OutMsgQueueImporter::get_proof_import(std::shared_ptr<CacheEntry> entry, std::vector<BlockIdExt> blocks,
+                                           block::ImportedMsgQueueLimits limits) {
+  if (!check_timeout(entry)) {
+    return;
+  }
+  td::actor::send_closure(
+      manager_, &ValidatorManager::send_get_out_msg_queue_proof_request, entry->dst_shard, blocks, limits,
+      [=, SelfId = actor_id(this), retry_after = td::Timestamp::in(0.5),
+       dst_shard = entry->dst_shard](td::Result<std::vector<td::Ref<OutMsgQueueProof>>> R) {
+        if (R.is_error()) {
+          LOG(DEBUG) << "Failed to get out msg queue for " << dst_shard.to_str() << ": " << R.move_as_error();
+          delay_action(
+              [=]() {
+                td::actor::send_closure(SelfId, &OutMsgQueueImporter::get_proof_import, entry, std::move(blocks),
+                                        limits);
+              },
+              retry_after);
+          return;
+        }
+        td::actor::send_closure(SelfId, &OutMsgQueueImporter::got_proof, entry, R.move_as_ok());
+      });
+}
+
+void OutMsgQueueImporter::got_proof(std::shared_ptr<CacheEntry> entry, std::vector<td::Ref<OutMsgQueueProof>> proofs) {
+  if (!check_timeout(entry)) {
+    return;
+  }
+  for (auto& p : proofs) {
+    entry->result[p->block_id_] = std::move(p);
+  }
+  CHECK(entry->pending > 0);
+  if (--entry->pending == 0) {
+    finish_query(entry);
+  }
+}
+
+void OutMsgQueueImporter::finish_query(std::shared_ptr<CacheEntry> entry) {
+  LOG(DEBUG) << "Done importing neighbor msg queues for shard " << entry->dst_shard.to_str() << ", "
+             << entry->blocks.size() << " blocks in " << entry->timer.elapsed() << "s";
+  entry->done = true;
+  alarm_timestamp().relax(entry->timeout = td::Timestamp::in(CACHE_TTL));
+  for (auto& p : entry->promises) {
+    p.first.set_result(entry->result);
+  }
+  entry->promises.clear();
+}
+
+bool OutMsgQueueImporter::check_timeout(std::shared_ptr<CacheEntry> entry) {
+  if (entry->timeout.is_in_past()) {
+    LOG(DEBUG) << "Aborting importing neighbor msg queues for shard " << entry->dst_shard.to_str() << ", "
+               << entry->blocks.size() << " blocks: timeout";
+    for (auto& p : entry->promises) {
+      p.first.set_error(td::Status::Error(ErrorCode::timeout, "timeout"));
+    }
+    entry->promises.clear();
+    auto it = cache_.find({entry->dst_shard, entry->blocks});
+    if (it != cache_.end() && it->second == entry) {
+      cache_.erase(it);
+    }
+    return false;
+  }
+  return true;
+}
+
+void OutMsgQueueImporter::alarm() {
+  auto it = cache_.begin();
+  while (it != cache_.end()) {
+    auto& promises = it->second->promises;
+    if (it->second->timeout.is_in_past()) {
+      if (!it->second->done) {
+        LOG(DEBUG) << "Aborting importing neighbor msg queues for shard " << it->second->dst_shard.to_str() << ", "
+                   << it->second->blocks.size() << " blocks: timeout";
+        for (auto& p : promises) {
+          p.first.set_error(td::Status::Error(ErrorCode::timeout, "timeout"));
+        }
+        promises.clear();
+      }
+      it = cache_.erase(it);
+      continue;
+    }
+    alarm_timestamp().relax(it->second->timeout);
+    size_t j = 0;
+    for (auto& p : promises) {
+      if (p.second.is_in_past()) {
+        p.first.set_error(td::Status::Error(ErrorCode::timeout, "timeout"));
+      } else {
+        alarm_timestamp().relax(p.second);
+        promises[j++] = std::move(p);
+      }
+    }
+    promises.resize(j);
+    ++it;
+  }
 }
 
 void BuildOutMsgQueueProof::abort_query(td::Status reason) {
   if (promise_) {
-    LOG(DEBUG) << "failed to build msg queue proof for " << block_id_.to_str() << ": " << reason;
+    LOG(DEBUG) << "failed to build msg queue proof to " << dst_shard_.to_str() << ": " << reason;
     promise_.set_error(
-        reason.move_as_error_prefix(PSTRING() << "failed to build msg queue proof for " << block_id_.to_str() << ": "));
+        reason.move_as_error_prefix(PSTRING() << "failed to build msg queue proof to " << dst_shard_.to_str() << ": "));
   }
   stop();
 }
 
 void BuildOutMsgQueueProof::start_up() {
-  ++pending;
-  td::actor::send_closure(manager_, &ValidatorManagerInterface::get_shard_state_from_db_short, block_id_,
-                          [SelfId = actor_id(this)](td::Result<Ref<ShardState>> R) {
-                            if (R.is_error()) {
-                              td::actor::send_closure(SelfId, &BuildOutMsgQueueProof::abort_query,
-                                                      R.move_as_error_prefix("failed to get shard state: "));
-                            } else {
-                              td::actor::send_closure(SelfId, &BuildOutMsgQueueProof::got_state_root,
-                                                      R.move_as_ok()->root_cell());
-                            }
-                          });
-  if (block_id_.seqno() != 0) {
+  for (size_t i = 0; i < blocks_.size(); ++i) {
+    BlockIdExt id = blocks_[i].id;
     ++pending;
-    td::actor::send_closure(manager_, &ValidatorManagerInterface::get_block_data_from_db_short, block_id_,
-                            [SelfId = actor_id(this)](td::Result<Ref<BlockData>> R) {
+    td::actor::send_closure(manager_, &ValidatorManagerInterface::get_shard_state_from_db_short, id,
+                            [SelfId = actor_id(this), i](td::Result<Ref<ShardState>> R) {
                               if (R.is_error()) {
                                 td::actor::send_closure(SelfId, &BuildOutMsgQueueProof::abort_query,
-                                                        R.move_as_error_prefix("failed to get block data: "));
+                                                        R.move_as_error_prefix("failed to get shard state: "));
                               } else {
-                                td::actor::send_closure(SelfId, &BuildOutMsgQueueProof::got_block_root,
+                                td::actor::send_closure(SelfId, &BuildOutMsgQueueProof::got_state_root, i,
                                                         R.move_as_ok()->root_cell());
                               }
                             });
+    if (id.seqno() != 0) {
+      ++pending;
+      td::actor::send_closure(manager_, &ValidatorManagerInterface::get_block_data_from_db_short, id,
+                              [SelfId = actor_id(this), i](td::Result<Ref<BlockData>> R) {
+                                if (R.is_error()) {
+                                  td::actor::send_closure(SelfId, &BuildOutMsgQueueProof::abort_query,
+                                                          R.move_as_error_prefix("failed to get block data: "));
+                                } else {
+                                  td::actor::send_closure(SelfId, &BuildOutMsgQueueProof::got_block_root, i,
+                                                          R.move_as_ok()->root_cell());
+                                }
+                              });
+    }
+  }
+  if (pending == 0) {
+    build_proof();
   }
 }
 
-void BuildOutMsgQueueProof::got_state_root(Ref<vm::Cell> root) {
-  state_root_ = std::move(root);
+void BuildOutMsgQueueProof::got_state_root(size_t i, Ref<vm::Cell> root) {
+  blocks_[i].state_root = std::move(root);
   if (--pending == 0) {
     build_proof();
   }
 }
 
-void BuildOutMsgQueueProof::got_block_root(Ref<vm::Cell> root) {
-  block_root_ = std::move(root);
+void BuildOutMsgQueueProof::got_block_root(size_t i, Ref<vm::Cell> root) {
+  blocks_[i].block_root = std::move(root);
   if (--pending == 0) {
     build_proof();
   }
 }
 
 void BuildOutMsgQueueProof::build_proof() {
-  auto result = OutMsgQueueProof::build(block_id_, dst_shard_, limits_, std::move(state_root_), std::move(block_root_));
+  auto result = OutMsgQueueProof::build(dst_shard_, std::move(blocks_), limits_);
   if (result.is_error()) {
     LOG(ERROR) << "Failed to build msg queue proof: " << result.error();
   }

validator/impl/out-msg-queue-proof.hpp

@@ -20,6 +20,8 @@
 #include "auto/tl/ton_api.h"
 #include "interfaces/out-msg-queue-proof.h"
 #include "td/actor/actor.h"
+#include "interfaces/shard.h"
+#include "validator.h"
 
 namespace ton {
 
@@ -29,83 +31,77 @@ using td::Ref;
 class ValidatorManager;
 class ValidatorManagerInterface;
 
-class WaitOutMsgQueueProof : public td::actor::Actor {
+class OutMsgQueueImporter : public td::actor::Actor {
  public:
-  WaitOutMsgQueueProof(BlockIdExt block_id, ShardIdFull dst_shard, block::ImportedMsgQueueLimits limits, bool local,
-                       td::uint32 priority, td::actor::ActorId<ValidatorManager> manager, td::Timestamp timeout,
-                       td::Promise<Ref<OutMsgQueueProof>> promise)
-      : block_id_(std::move(block_id))
-      , dst_shard_(dst_shard)
-      , limits_(limits)
-      , local_(local)
-      , priority_(priority)
-      , manager_(manager)
-      , timeout_(timeout)
-      , promise_(std::move(promise)) {
+  OutMsgQueueImporter(td::actor::ActorId<ValidatorManager> manager, td::Ref<ValidatorManagerOptions> opts,
+                      td::Ref<MasterchainState> last_masterchain_state)
+      : manager_(manager), opts_(opts), last_masterchain_state_(last_masterchain_state) {
   }
 
-  void update_timeout(td::Timestamp timeout, td::uint32 priority) {
-    timeout_ = timeout;
-    alarm_timestamp() = timeout_;
-    priority_ = priority;
+  void new_masterchain_block_notification(td::Ref<MasterchainState> state, std::set<ShardIdFull> collating_shards);
+  void get_neighbor_msg_queue_proofs(ShardIdFull dst_shard, std::vector<BlockIdExt> blocks, td::Timestamp timeout,
+                                     td::Promise<std::map<BlockIdExt, td::Ref<OutMsgQueueProof>>> promise);
+
+  void update_options(td::Ref<ValidatorManagerOptions> opts) {
+    opts_ = std::move(opts);
   }
 
-  void abort_query(td::Status reason);
-  void finish_query(Ref<OutMsgQueueProof> result);
   void alarm() override;
 
-  void start_up() override;
-
-  void run_local();
-  void got_state_root(Ref<vm::Cell> root);
-  void got_block_root(Ref<vm::Cell> root);
-  void run_local_cont();
-
-  void run_net();
-
  private:
-  BlockIdExt block_id_;
-  ShardIdFull dst_shard_;
-  block::ImportedMsgQueueLimits limits_;
-  bool local_;
-  td::uint32 priority_;
-
   td::actor::ActorId<ValidatorManager> manager_;
-  td::Timestamp timeout_;
-  td::Promise<Ref<OutMsgQueueProof>> promise_;
+  td::Ref<ValidatorManagerOptions> opts_;
+  td::Ref<MasterchainState> last_masterchain_state_;
 
-  Ref<vm::Cell> state_root_, block_root_;
-  unsigned pending = 0;
+  struct CacheEntry {
+    ShardIdFull dst_shard;
+    std::vector<BlockIdExt> blocks;
+    bool done = false;
+    std::map<BlockIdExt, td::Ref<OutMsgQueueProof>> result;
+    std::vector<std::pair<td::Promise<std::map<BlockIdExt, td::Ref<OutMsgQueueProof>>>, td::Timestamp>> promises;
+    td::Timestamp timeout = td::Timestamp::never();
+    td::Timer timer;
+    size_t pending = 0;
+  };
+  std::map<std::pair<ShardIdFull, std::vector<BlockIdExt>>, std::shared_ptr<CacheEntry>> cache_;
+
+  void get_proof_local(std::shared_ptr<CacheEntry> entry, BlockIdExt block);
+  void get_proof_import(std::shared_ptr<CacheEntry> entry, std::vector<BlockIdExt> blocks,
+                        block::ImportedMsgQueueLimits limits);
+  void got_proof(std::shared_ptr<CacheEntry> entry, std::vector<td::Ref<OutMsgQueueProof>> proofs);
+  void finish_query(std::shared_ptr<CacheEntry> entry);
+  bool check_timeout(std::shared_ptr<CacheEntry> entry);
+
+  constexpr static const double CACHE_TTL = 30.0;
 };
 
 class BuildOutMsgQueueProof : public td::actor::Actor {
  public:
-  BuildOutMsgQueueProof(BlockIdExt block_id, ShardIdFull dst_shard, block::ImportedMsgQueueLimits limits,
+  BuildOutMsgQueueProof(ShardIdFull dst_shard, std::vector<BlockIdExt> blocks, block::ImportedMsgQueueLimits limits,
                         td::actor::ActorId<ValidatorManagerInterface> manager,
                         td::Promise<tl_object_ptr<ton_api::tonNode_outMsgQueueProof>> promise)
-      : block_id_(std::move(block_id))
-      , dst_shard_(dst_shard)
-      , limits_(limits)
-      , manager_(manager)
-      , promise_(std::move(promise)) {
+      : dst_shard_(dst_shard), limits_(limits), manager_(manager), promise_(std::move(promise)) {
+    blocks_.resize(blocks.size());
+    for (size_t i = 0; i < blocks_.size(); ++i) {
+      blocks_[i].id = blocks[i];
+    }
   }
 
   void abort_query(td::Status reason);
   void start_up() override;
-  void got_state_root(Ref<vm::Cell> root);
-  void got_block_root(Ref<vm::Cell> root);
+  void got_state_root(size_t i, Ref<vm::Cell> root);
+  void got_block_root(size_t i, Ref<vm::Cell> root);
   void build_proof();
 
  private:
-  BlockIdExt block_id_;
   ShardIdFull dst_shard_;
+  std::vector<OutMsgQueueProof::OneBlock> blocks_;
   block::ImportedMsgQueueLimits limits_;
 
   td::actor::ActorId<ValidatorManagerInterface> manager_;
   td::Promise<tl_object_ptr<ton_api::tonNode_outMsgQueueProof>> promise_;
 
-  Ref<vm::Cell> state_root_, block_root_;
-  unsigned pending = 0;
+  size_t pending = 0;
 };
 
 }  // namespace validator