Support optimistic collation/validation

Author: SpyCheese

tdutils/td/utils/Status.h

@@ -74,7 +74,7 @@
 #define TRY_RESULT_ASSIGN(name, result) TRY_RESULT_IMPL(TD_CONCAT(r_response, __LINE__), name, result)
 
 #define TRY_RESULT_PROMISE_ASSIGN(promise_name, name, result) \
-  TRY_RESULT_PROMISE_IMPL(promise_name, TD_CONCAT(TD_CONCAT(r_, name), __LINE__), name, result)
+  TRY_RESULT_PROMISE_IMPL(promise_name, TD_CONCAT(r_response, __LINE__), name, result)
 
 #define TRY_RESULT_PREFIX(name, result, prefix) \
   TRY_RESULT_PREFIX_IMPL(TD_CONCAT(TD_CONCAT(r_, name), __LINE__), auto name, result, prefix)
@@ -86,7 +86,7 @@
   TRY_RESULT_PROMISE_PREFIX_IMPL(promise_name, TD_CONCAT(TD_CONCAT(r_, name), __LINE__), auto name, result, prefix)
 
 #define TRY_RESULT_PROMISE_PREFIX_ASSIGN(promise_name, name, result, prefix) \
-  TRY_RESULT_PROMISE_PREFIX_IMPL(promise_name, TD_CONCAT(TD_CONCAT(r_, name), __LINE__), name, result, prefix)
+  TRY_RESULT_PROMISE_PREFIX_IMPL(promise_name, TD_CONCAT(r_response, __LINE__), name, result, prefix)
 
 #define TRY_RESULT_IMPL(r_name, name, result) \
   auto r_name = (result);                     \

validator-session/validator-session.cpp

@@ -1578,18 +1578,18 @@ void ValidatorSessionImpl::generate_block_optimistic(td::uint32 cur_round,
   if (!stat) {
     return;
   }
-  callback_->generate_block_optimistic(BlockSourceInfo{description().get_source_public_key(local_idx()),
-                                                       BlockCandidatePriority{cur_round + 1, cur_round + 1, 0}},
-                                       block->data_.clone(), block->root_hash_, stat->block_id.file_hash,
-                                       [=, SelfId = actor_id(this)](td::Result<GeneratedCandidate> R) {
-                                         if (R.is_error()) {
-                                           LOG(DEBUG) << "Optimistic generation error: " << R.move_as_error();
-                                           return;
-                                         }
-                                         td::actor::send_closure(SelfId,
-                                                                 &ValidatorSessionImpl::generated_optimistic_candidate,
-                                                                 cur_round + 1, R.move_as_ok(), prev_candidate_id);
-                                       });
+  callback_->generate_block_optimistic(
+      BlockSourceInfo{description().get_source_public_key(local_idx()),
+                      BlockCandidatePriority{cur_round + 1, cur_round + 1, 0}},
+      block->data_.clone(), block->collated_data_.clone(), block->root_hash_, stat->block_id.file_hash,
+      [=, SelfId = actor_id(this)](td::Result<GeneratedCandidate> R) {
+        if (R.is_error()) {
+          LOG(DEBUG) << "Optimistic generation error: " << R.move_as_error();
+          return;
+        }
+        td::actor::send_closure(SelfId, &ValidatorSessionImpl::generated_optimistic_candidate, cur_round + 1,
+                                R.move_as_ok(), prev_candidate_id);
+      });
 }
 
 void ValidatorSessionImpl::generated_optimistic_candidate(td::uint32 round, GeneratedCandidate candidate,

validator-session/validator-session.h

@@ -95,8 +95,8 @@ class ValidatorSession : public td::actor::Actor {
                                         ValidatorSessionCollatedDataFileHash collated_data_file_hash,
                                         td::Promise<BlockCandidate> promise) = 0;
     virtual void generate_block_optimistic(BlockSourceInfo source_info, td::BufferSlice prev_block,
-                                           RootHash prev_root_hash, FileHash prev_file_hash,
-                                           td::Promise<GeneratedCandidate> promise) {
+                                           td::BufferSlice prev_collated_data, RootHash prev_root_hash,
+                                           FileHash prev_file_hash, td::Promise<GeneratedCandidate> promise) {
     }
     virtual void on_optimistic_candidate(BlockSourceInfo source_info, ValidatorSessionRootHash root_hash,
                                          td::BufferSlice data, td::BufferSlice collated_data, PublicKey prev_source,

validator/collation-manager.cpp

@@ -159,7 +159,7 @@ void CollationManager::collate_shard_block(CollateParams params, BlockCandidateP
     next_block_id.seqno = std::max(next_block_id.seqno, p.seqno() + 1);
   }
 
-  td::Promise<BlockCandidate> P = [=, SelfId = actor_id(this), promise = std::move(promise),
+  td::Promise<BlockCandidate> P = [=, SelfId = actor_id(this), promise = std::move(promise), params = params.clone(),
                                    retry_at = td::Timestamp::in(0.5)](td::Result<BlockCandidate> R) mutable {
     if (R.is_ok()) {
       promise.set_value(GeneratedCandidate{.candidate = R.move_as_ok(),
@@ -178,9 +178,9 @@ void CollationManager::collate_shard_block(CollateParams params, BlockCandidateP
       return;
     }
     delay_action(
-        [=, promise = std::move(promise)]() mutable {
-          td::actor::send_closure(SelfId, &CollationManager::collate_shard_block, params, priority, max_answer_size,
-                                  cancellation_token, std::move(promise), timeout);
+        [=, promise = std::move(promise), params = std::move(params)]() mutable {
+          td::actor::send_closure(SelfId, &CollationManager::collate_shard_block, std::move(params), priority,
+                                  max_answer_size, cancellation_token, std::move(promise), timeout);
         },
         retry_at);
   };
@@ -203,7 +203,7 @@ void CollationManager::collate_shard_block(CollateParams params, BlockCandidateP
   LOG(INFO) << "sending collate query for " << next_block_id.to_str() << ": send to #" << selected_idx << "("
             << selected_collator << ")";
 
-  td::Promise<td::BufferSlice> P2 = [=, SelfId = actor_id(this), P = std::move(P),
+  td::Promise<td::BufferSlice> P2 = [=, SelfId = actor_id(this), P = std::move(P), creator = params.creator,
                                      timer = td::Timer()](td::Result<td::BufferSlice> R) mutable {
     TRY_RESULT_PROMISE_PREFIX(P, data, std::move(R), "rldp query failed: ");
     auto r_error = fetch_tl_object<ton_api::collatorNode_error>(data, true);
@@ -214,7 +214,7 @@ void CollationManager::collate_shard_block(CollateParams params, BlockCandidateP
     }
     TRY_RESULT_PROMISE(P, f, fetch_tl_object<ton_api::collatorNode_Candidate>(data, true));
     TRY_RESULT_PROMISE(P, candidate, deserialize_candidate(std::move(f), td::narrow_cast<int>(max_answer_size)));
-    if (candidate.pubkey.as_bits256() != params.creator.as_bits256()) {
+    if (candidate.pubkey.as_bits256() != creator.as_bits256()) {
       P.set_error(td::Status::Error("collate query: block candidate source mismatch"));
       return;
     }
@@ -231,6 +231,7 @@ void CollationManager::collate_shard_block(CollateParams params, BlockCandidateP
     BlockIdExt prev_block_id = params.optimistic_prev_block->block_id();
     auto& entry = optimistic_prev_cache_[prev_block_id];
     entry.block_data = params.optimistic_prev_block->data().clone();
+    entry.collated_data = params.optimistic_prev_collated_data.clone();
     ++entry.refcnt;
     P2 = [this, SelfId = actor_id(this), prev_block_id, P2 = std::move(P2)](td::Result<td::BufferSlice> R) mutable {
       P2.set_result(std::move(R));
@@ -479,18 +480,22 @@ void CollationManager::receive_query(adnl::AdnlNodeIdShort src, td::BufferSlice
   }
   TRY_RESULT_PROMISE(promise, query, fetch_tl_object<ton_api::collatorNode_requestBlockCallback>(data, true));
   BlockIdExt block_id = create_block_id(query->block_id_);
+  bool with_collated_data = query->flags_ & 1;
   auto it = optimistic_prev_cache_.find(block_id);
   if (it == optimistic_prev_cache_.end()) {
-    LOG(INFO) << "collatorNode.requestBlockCallback from " << src << " block " << block_id.to_str() << " : not found";
+    LOG(INFO) << "collatorNode.requestBlockCallback from " << src << " block " << block_id.to_str() << " ("
+              << (with_collated_data ? "with" : "no") << " cdata) : not found";
     promise.set_error(td::Status::Error("block not found"));
     return;
   }
-  LOG(INFO) << "collatorNode.requestBlockCallback from " << src << " block " << block_id.to_str() << " : OK";
-  promise.set_value(
-      serialize_tl_object(serialize_candidate(BlockCandidate(Ed25519_PublicKey{td::Bits256::zero()}, block_id,
-                                                             td::Bits256::zero(), it->second.block_data.clone(), {}),
-                                              true),
-                          true));
+  LOG(INFO) << "collatorNode.requestBlockCallback from " << src << " block " << block_id.to_str() << " ("
+            << (with_collated_data ? "with" : "no") << " cdata) : OK";
+  promise.set_value(serialize_tl_object(
+      serialize_candidate(BlockCandidate(Ed25519_PublicKey{td::Bits256::zero()}, block_id, td::Bits256::zero(),
+                                         it->second.block_data.clone(),
+                                         with_collated_data ? it->second.collated_data.clone() : td::BufferSlice{}),
+                          true),
+      true));
 }
 
 }  // namespace ton::validator

validator/collation-manager.hpp

@@ -96,6 +96,7 @@ class CollationManager : public td::actor::Actor {
 
   struct OptimisticPrevCache {
     td::BufferSlice block_data;
+    td::BufferSlice collated_data;
     size_t refcnt = 0;
   };
   std::map<BlockIdExt, OptimisticPrevCache> optimistic_prev_cache_;

validator/collator-node/collator-node-session.cpp

@@ -62,7 +62,7 @@ void CollatorNodeSession::start_up() {
     collated_data_deduplicator_ = std::make_shared<CollatedDataDeduplicator>();
   }
   if (can_generate_) {
-    generate_block(prev_, {}, {}, td::Timestamp::in(10.0), [](td::Result<BlockCandidate>) {});
+    generate_block(prev_, {}, {}, {}, td::Timestamp::in(10.0), [](td::Result<BlockCandidate>) {});
   }
 }
 
@@ -72,6 +72,11 @@ void CollatorNodeSession::tear_down() {
   for (auto& [_, entry] : cache_) {
     entry->cancel(td::Status::Error("validator session finished"));
   }
+  for (auto& [_, vec] : collated_data_merged_waiters_) {
+    for (auto& [promise, _] : vec) {
+      promise.set_error(td::Status::Error("validator session finished"));
+    }
+  }
 }
 
 void CollatorNodeSession::new_shard_block_accepted(BlockIdExt block_id, bool can_generate) {
@@ -114,7 +119,7 @@ void CollatorNodeSession::new_shard_block_accepted(BlockIdExt block_id, bool can
   try_merge_collated_data(block_id);
 
   if (can_generate_) {
-    generate_block(prev_, {}, {}, td::Timestamp::in(10.0), [](td::Result<BlockCandidate>) {});
+    generate_block(prev_, {}, {}, {}, td::Timestamp::in(10.0), [](td::Result<BlockCandidate>) {});
   }
 }
 
@@ -143,7 +148,8 @@ void CollatorNodeSession::update_masterchain_config(td::Ref<MasterchainState> st
 
 void CollatorNodeSession::generate_block(std::vector<BlockIdExt> prev_blocks,
                                          td::optional<BlockCandidatePriority> o_priority,
-                                         td::Ref<BlockData> o_optimistic_prev_block, td::Timestamp timeout,
+                                         td::Ref<BlockData> o_optimistic_prev_block,
+                                         td::BufferSlice o_optimistic_prev_collated_data, td::Timestamp timeout,
                                          td::Promise<BlockCandidate> promise) {
   bool is_external = !o_priority;
   bool is_optimistic = o_optimistic_prev_block.not_null();
@@ -242,6 +248,7 @@ void CollatorNodeSession::generate_block(std::vector<BlockIdExt> prev_blocks,
                        .collator_node_id = local_id_,
                        .skip_store_candidate = true,
                        .optimistic_prev_block = o_optimistic_prev_block,
+                       .optimistic_prev_collated_data = std::move(o_optimistic_prev_collated_data),
                        .collated_data_deduplicator = collated_data_deduplicator_};
   auto token = cache_entry->cancellation_token_source.get_cancellation_token();
   wait_collated_data_merged(
@@ -295,26 +302,33 @@ void CollatorNodeSession::process_request(adnl::AdnlNodeIdShort src, std::vector
     if (it == cache_.end() || it->second->started) {
       BlockIdExt prev_block = prev_blocks[0];
       td::actor::send_closure(
-          manager_, &ValidatorManager::get_candidate_data_by_block_id_from_db, prev_block,
-          [=, SelfId = actor_id(this), promise = std::move(promise)](td::Result<td::BufferSlice> R) mutable {
+          manager_, &ValidatorManager::get_block_candidate_by_block_id_from_db, prev_block,
+          [=, SelfId = actor_id(this), promise = std::move(promise)](td::Result<BlockCandidate> R) mutable {
+            td::Result<std::pair<td::BufferSlice, td::BufferSlice>> res;
+            if (R.is_error()) {
+              res = R.move_as_error();
+            } else {
+              BlockCandidate c = R.move_as_ok();
+              res = std::make_pair(std::move(c.data), std::move(c.collated_data));
+            }
             td::actor::send_closure(SelfId, &CollatorNodeSession::process_request_optimistic_cont, src, prev_block,
-                                    priority, timeout, std::move(promise), std::move(R));
+                                    priority, timeout, std::move(promise), std::move(res));
           });
       return;
     }
   }
-  generate_block(std::move(prev_blocks), priority, {}, timeout, std::move(promise));
+  generate_block(std::move(prev_blocks), priority, {}, {}, timeout, std::move(promise));
 }
 
-void CollatorNodeSession::process_request_optimistic_cont(adnl::AdnlNodeIdShort src, BlockIdExt prev_block_id,
-                                                          BlockCandidatePriority priority, td::Timestamp timeout,
-                                                          td::Promise<BlockCandidate> promise,
-                                                          td::Result<td::BufferSlice> prev_block_data) {
-  if (prev_block_data.is_ok()) {
-    TRY_RESULT_PROMISE_PREFIX(promise, prev_block, create_block(prev_block_id, prev_block_data.move_as_ok()),
+void CollatorNodeSession::process_request_optimistic_cont(
+    adnl::AdnlNodeIdShort src, BlockIdExt prev_block_id, BlockCandidatePriority priority, td::Timestamp timeout,
+    td::Promise<BlockCandidate> promise, td::Result<std::pair<td::BufferSlice, td::BufferSlice>> prev_candidate) {
+  if (prev_candidate.is_ok()) {
+    auto [prev_block_data, prev_collated_data] = prev_candidate.move_as_ok();
+    TRY_RESULT_PROMISE_PREFIX(promise, prev_block, create_block(prev_block_id, std::move(prev_block_data)),
                               "invalid prev block data in db: ");
     LOG(INFO) << "got prev block from db for optimistic collation: " << prev_block_id.to_str();
-    generate_block({prev_block_id}, priority, prev_block, timeout, std::move(promise));
+    generate_block({prev_block_id}, priority, prev_block, std::move(prev_collated_data), timeout, std::move(promise));
     return;
   }
   td::actor::send_closure(
@@ -324,7 +338,8 @@ void CollatorNodeSession::process_request_optimistic_cont(adnl::AdnlNodeIdShort
                                 timeout, std::move(promise), std::move(R));
       },
       timeout,
-      create_serialize_tl_object<ton_api::collatorNode_requestBlockCallback>(0, create_tl_block_id(prev_block_id)),
+      create_serialize_tl_object<ton_api::collatorNode_requestBlockCallback>(merge_collated_data_enabled_ ? 1 : 0,
+                                                                             create_tl_block_id(prev_block_id)),
       max_candidate_size_);
 }
 
@@ -335,13 +350,29 @@ void CollatorNodeSession::process_request_optimistic_cont2(BlockIdExt prev_block
                             "failed to download prev block data for optimistic collation: ");
   TRY_RESULT_PROMISE_PREFIX(promise, f, fetch_tl_object<ton_api::collatorNode_Candidate>(response, true),
                             "failed to download prev block data for optimistic collation: ");
-  TRY_RESULT_PROMISE_PREFIX(promise, candidate,
-                            deserialize_candidate(std::move(f), max_candidate_size_),
+  TRY_RESULT_PROMISE_PREFIX(promise, candidate, deserialize_candidate(std::move(f), max_candidate_size_),
                             "failed to download prev block data for optimistic collation: ");
   TRY_RESULT_PROMISE_PREFIX(promise, prev_block, create_block(prev_block_id, std::move(candidate.data)),
                             "invalid prev block data from validator: ");
+  if (merge_collated_data_enabled_) {
+    block::gen::Block::Record rec;
+    block::gen::BlockInfo::Record info;
+    if (!block::gen::unpack_cell(prev_block->root_cell(), rec) || !block::gen::unpack_cell(rec.info, info)) {
+      promise.set_error(td::Status::Error("failed to unpack prev block header"));
+      return;
+    }
+    if (info.flags & 2) {
+      FileHash stored_collated_data_hash;
+      info.collated_data_hash->prefetch_bits_to(stored_collated_data_hash);
+      if (stored_collated_data_hash != candidate.collated_file_hash) {
+        promise.set_error(td::Status::Error("collated data hash mismatch"));
+        return;
+      }
+    }
+  }
   LOG(INFO) << "got prev block from validator for optimistic collation: " << prev_block_id.to_str();
-  generate_block({prev_block_id}, priority, prev_block, timeout, std::move(promise));
+  generate_block({prev_block_id}, priority, prev_block, std::move(candidate.collated_data), timeout,
+                 std::move(promise));
 }
 
 void CollatorNodeSession::CacheEntry::cancel(td::Status reason) {

validator/collator-node/collator-node-session.hpp

@@ -82,14 +82,14 @@ class CollatorNodeSession : public td::actor::Actor {
   td::uint32 max_candidate_size_ = 0;
 
   void generate_block(std::vector<BlockIdExt> prev_blocks, td::optional<BlockCandidatePriority> o_priority,
-                      td::Ref<BlockData> o_optimistic_prev_block, td::Timestamp timeout,
-                      td::Promise<BlockCandidate> promise);
+                      td::Ref<BlockData> o_optimistic_prev_block, td::BufferSlice o_optimistic_prev_collated_data,
+                      td::Timestamp timeout, td::Promise<BlockCandidate> promise);
   void process_result(std::shared_ptr<CacheEntry> cache_entry, td::Result<BlockCandidate> R);
 
   void process_request_optimistic_cont(adnl::AdnlNodeIdShort src, BlockIdExt prev_block_id,
                                        BlockCandidatePriority priority, td::Timestamp timeout,
                                        td::Promise<BlockCandidate> promise,
-                                       td::Result<td::BufferSlice> prev_block_data);
+                                       td::Result<std::pair<td::BufferSlice, td::BufferSlice>> prev_candidate);
   void process_request_optimistic_cont2(BlockIdExt prev_block_id, BlockCandidatePriority priority,
                                         td::Timestamp timeout, td::Promise<BlockCandidate> promise,
                                         td::Result<td::BufferSlice> R);

validator/collator-node/collator-node.cpp

@@ -193,7 +193,8 @@ void CollatorNode::new_masterchain_block_notification(td::Ref<MasterchainState>
         }
       }
       for (BlockCandidate& candidate : future_group.pending_candidate_broadcasts) {
-        td::actor::send_closure(it->second.actor, &CollatorNodeSession::on_block_candidate_broadcast, std::move(candidate));
+        td::actor::send_closure(it->second.actor, &CollatorNodeSession::on_block_candidate_broadcast,
+                                std::move(candidate));
       }
       for (auto& promise : future_group.promises) {
         promise.set_value(td::Unit());