Write collation stats to session stats, add collator options for collated data

tl/generate/scheme/ton_api.tl

@@ -669,7 +669,8 @@ engine.validator.collatorOptions
   deferring_enabled:Bool defer_messages_after:int defer_out_queue_size_limit:long
   dispatch_phase_2_max_total:int dispatch_phase_3_max_total:int
   dispatch_phase_2_max_per_initiator:int dispatch_phase_3_max_per_initiator:int
-  whitelist:(vector string) prioritylist:(vector string) = engine.validator.CollatorOptions;
+  whitelist:(vector string) prioritylist:(vector string)
+  force_full_collated_data:Bool ignore_collated_data_limits:Bool = engine.validator.CollatorOptions;
 
 engine.validator.collatorsList.collator adnl_id:int256 = engine.validator.collatorsList.Collator;
 engine.validator.collatorsList.shard shard_id:tonNode.shardId collators:(vector engine.validator.collatorsList.collator)
@@ -874,8 +875,12 @@ http.server.config dhs:(vector http.server.dnsEntry) local_hosts:(vector http.se
 
 ---types---
 
-validatorSession.collationStats bytes:int gas:int lt_delta:int cat_bytes:int cat_gas:int cat_lt_delta:int
+validatorSession.collationStats actual_bytes:int actual_collated_data_bytes:int
-  limits_log:string ext_msgs_total:int ext_msgs_filtered:int ext_msgs_accepted:int ext_msgs_rejected:int = validadorSession.CollationStats;
+  bytes:int gas:int lt_delta:int collated_data_bytes:int
+  cat_bytes:int cat_gas:int cat_lt_delta:int cat_collated_data_bytes:int
+  limits_log:string ext_msgs_total:int ext_msgs_filtered:int ext_msgs_accepted:int ext_msgs_rejected:int
+  work_time:double cpu_work_time:double
+  serialized_size:int serialized_size_no_collated_data:int = validadorSession.CollationStats;
 
 validatorSession.statsProducer id:int256 candidate_id:int256 block_status:int root_hash:int256 file_hash:int256
   comment:string block_timestamp:double is_accepted:Bool is_ours:Bool got_submit_at:double
@@ -896,6 +901,8 @@ validatorSession.stats success:Bool id:tonNode.blockIdExt timestamp:double self:
     signatures:int signatures_weight:long approve_signatures:int approve_signatures_weight:long
     first_round:int rounds:(vector validatorSession.statsRound) = validatorSession.Stats;
 
+validatorSession.statsCollatedBlock timestamp:double id:tonNode.blockIdExt stats:validatorSession.collationStats = validatorSession.StatsCollatedBlock;
+
 collatorNode.candidate source:PublicKey id:tonNode.blockIdExt data:bytes collated_data:bytes = collatorNode.Candidate;
 collatorNode.compressedCandidate flags:# source:PublicKey id:tonNode.blockIdExt decompressed_size:int data:bytes = collatorNode.Candidate;
 collatorNode.pong flags:# = collatorNode.Pong;

validator-engine/validator-engine.cpp

@@ -2707,6 +2707,8 @@ static td::Result<td::Ref<ton::validator::CollatorOptions>> parse_collator_optio
   f.dispatch_phase_2_max_per_initiator_ = opts.dispatch_phase_2_max_per_initiator;
   f.dispatch_phase_3_max_per_initiator_ =
       opts.dispatch_phase_3_max_per_initiator ? opts.dispatch_phase_3_max_per_initiator.value() : -1;
+  f.force_full_collated_data_ = false;
+  f.ignore_collated_data_limits_ = false;
 
   TRY_RESULT_PREFIX(json, td::json_decode(json_str), "failed to parse json: ");
   TRY_STATUS_PREFIX(ton::ton_api::from_json(f, json.get_object()), "json does not fit TL scheme: ");
@@ -2746,6 +2748,8 @@ static td::Result<td::Ref<ton::validator::CollatorOptions>> parse_collator_optio
     TRY_RESULT(addr, block::StdAddress::parse(s));
     opts.prioritylist.emplace(addr.workchain, addr.addr);
   }
+  opts.force_full_collated_data = f.force_full_collated_data_;
+  opts.ignore_collated_data_limits = f.ignore_collated_data_limits_;
 
   return ref;
 }

validator/collator-node.cpp

@@ -431,7 +431,8 @@ void CollatorNode::generate_block(ShardIdFull shard, CatchainSeqno cc_seqno, std
                   << ", time=" << timer.elapsed() << ": " << (R.is_ok() ? "OK" : R.error().to_string());
         td::actor::send_closure(SelfId, &CollatorNode::process_result, cache_entry, std::move(R));
       },
-      cache_entry->cancellation_token_source.get_cancellation_token(), CollateMode::skip_store_candidate);
+      cache_entry->cancellation_token_source.get_cancellation_token(),
+      CollateMode::skip_store_candidate | CollateMode::from_collator_node);
 }
 
 void CollatorNode::process_result(std::shared_ptr<CacheEntry> cache_entry, td::Result<BlockCandidate> R) {

validator/fabric.h

@@ -27,7 +27,7 @@ namespace ton {
 namespace validator {
 
 enum ValidateMode { fake = 1 };
-enum CollateMode { skip_store_candidate = 1 };
+enum CollateMode { skip_store_candidate = 1, from_collator_node = 2 };
 
 td::actor::ActorOwn<Db> create_db_actor(td::actor::ActorId<ValidatorManager> manager, std::string db_root_,
                                         td::Ref<ValidatorManagerOptions> opts);

validator/impl/collator.cpp

@@ -16,6 +16,7 @@
 
     Copyright 2017-2020 Telegram Systems LLP
 */
+#include "candidate-serializer.h"
 #include "collator-impl.h"
 #include "vm/boc.h"
 #include "td/db/utils/BlobView.h"
@@ -77,7 +78,7 @@ static inline bool dbg(int c) {
  * @param timeout The timeout for the collator.
  * @param promise The promise to return the result.
  * @param cancellation_token Token to cancel collation.
- * @param mode +1 - skip storing candidate to disk.
+ * @param mode +1 - skip storing candidate to disk, +2 - called from CollatorNode.
  * @param attempt_idx The index of the attempt, starting from 0. On later attempts collator decreases block limits and skips some steps.
  */
 Collator::Collator(ShardIdFull shard, bool is_hardfork, BlockIdExt min_masterchain_block_id,
@@ -730,7 +731,7 @@ bool Collator::unpack_last_mc_state() {
   store_out_msg_queue_size_ = config_->has_capability(ton::capStoreOutMsgQueueSize);
   msg_metadata_enabled_ = config_->has_capability(ton::capMsgMetadata);
   deferring_messages_enabled_ = config_->has_capability(ton::capDeferMessages);
-  full_collated_data_ = config_->has_capability(capFullCollatedData);
+  full_collated_data_ = config_->has_capability(capFullCollatedData) || collator_opts_->force_full_collated_data;
   LOG(DEBUG) << "full_collated_data is " << full_collated_data_;
   shard_conf_ = std::make_unique<block::ShardConfig>(*config_);
   prev_key_block_exists_ = config_->get_last_key_block(prev_key_block_, prev_key_block_lt_);
@@ -751,15 +752,24 @@ bool Collator::unpack_last_mc_state() {
     LOG(INFO) << "Attempt #3: bytes, gas limits /= 2";
     block_limits_->bytes.multiply_by(0.5);
     block_limits_->gas.multiply_by(0.5);
+    block_limits_->collated_data.multiply_by(0.5);
   } else if (attempt_idx_ == 4) {
     LOG(INFO) << "Attempt #4: bytes, gas limits /= 4";
     block_limits_->bytes.multiply_by(0.25);
     block_limits_->gas.multiply_by(0.25);
+    block_limits_->collated_data.multiply_by(0.25);
+  }
+  if (collator_opts_->ignore_collated_data_limits) {
+    block_limits_->collated_data = block::ParamLimits{1 << 30, 1 << 30, 1 << 30};
   }
   LOG(DEBUG) << "block limits: bytes [" << block_limits_->bytes.underload() << ", " << block_limits_->bytes.soft()
              << ", " << block_limits_->bytes.hard() << "]";
   LOG(DEBUG) << "block limits: gas [" << block_limits_->gas.underload() << ", " << block_limits_->gas.soft() << ", "
              << block_limits_->gas.hard() << "]";
+  LOG(DEBUG) << "block limits: lt_delta [" << block_limits_->lt_delta.underload() << ", "
+             << block_limits_->lt_delta.soft() << ", " << block_limits_->lt_delta.hard() << "]";
+  LOG(DEBUG) << "block limits: collated_data_bytes [" << block_limits_->collated_data.underload() << ", "
+             << block_limits_->collated_data.soft() << ", " << block_limits_->collated_data.hard() << "]";
   if (config_->has_capabilities() && (config_->get_capabilities() & ~supported_capabilities())) {
     LOG(ERROR) << "block generation capabilities " << config_->get_capabilities()
                << " have been enabled in global configuration, but we support only " << supported_capabilities()
@@ -3678,6 +3688,10 @@ static std::string block_full_comment(const block::BlockLimitStatus& block_limit
   if (!block_limit_status.limits.lt_delta.fits(cls, lt_delta)) {
     return PSTRING() << "block_full lt_delta " << lt_delta;
   }
+  auto collated_data_bytes = block_limit_status.collated_data_stat.estimate_proof_size();
+  if (!block_limit_status.limits.collated_data.fits(cls, collated_data_bytes)) {
+    return PSTRING() << "block_full collated_data " << collated_data_bytes;
+  }
   return "";
 }
 
@@ -5810,7 +5824,8 @@ bool Collator::create_block_candidate() {
                                  << ") exceeds the limit in consensus config (" << consensus_config.max_block_size
                                  << ")");
   }
-  if (block_candidate->collated_data.size() > consensus_config.max_collated_data_size) {
+  if (block_candidate->collated_data.size() > consensus_config.max_collated_data_size &&
+      !collator_opts_->ignore_collated_data_limits) {
     return fatal_error(PSTRING() << "collated data size (" << block_candidate->collated_data.size()
                                  << ") exceeds the limit in consensus config ("
                                  << consensus_config.max_collated_data_size << ")");
@@ -5838,14 +5853,31 @@ bool Collator::create_block_candidate() {
   double work_time = work_timer_.elapsed();
   double cpu_work_time = cpu_work_timer_.elapsed();
   LOG(WARNING) << "Collate query work time = " << work_time << "s, cpu time = " << cpu_work_time << "s";
-  stats_.bytes = block_limit_status_->estimate_block_size();
+  stats_.actual_bytes = block_candidate->data.size();
+  stats_.actual_collated_data_bytes = block_candidate->collated_data.size();
+  stats_.estimated_bytes = block_limit_status_->estimate_block_size();
   stats_.gas = block_limit_status_->gas_used;
   stats_.lt_delta = block_limit_status_->cur_lt - block_limit_status_->limits.start_lt;
-  stats_.cat_bytes = block_limit_status_->limits.classify_size(stats_.bytes);
+  stats_.estimated_collated_data_bytes = block_limit_status_->collated_data_stat.estimate_proof_size();
+  stats_.cat_bytes = block_limit_status_->limits.classify_size(stats_.estimated_bytes);
   stats_.cat_gas = block_limit_status_->limits.classify_gas(stats_.gas);
   stats_.cat_lt_delta = block_limit_status_->limits.classify_lt(block_limit_status_->cur_lt);
-  td::actor::send_closure(manager, &ValidatorManager::record_collate_query_stats, block_candidate->id, work_time,
+  stats_.cat_collated_data_bytes =
-                          cpu_work_time, std::move(stats_));
+      block_limit_status_->limits.classify_collated_data_size(stats_.estimated_collated_data_bytes);
+  stats_.work_time = work_time;
+  stats_.cpu_work_time = cpu_work_time;
+
+  // TODO: remove this later (currently needed to collect stats)
+  if (mode_ & CollateMode::from_collator_node) {
+    size_t d;
+    stats_.serialized_size =
+        validatorsession::compress_candidate_data(block_candidate->data, block_candidate->collated_data, d).ok().size();
+    stats_.serialized_size_no_collated_data =
+        validatorsession::compress_candidate_data(block_candidate->data, td::Slice{}, d).ok().size();
+  }
+
+  td::actor::send_closure(manager, &ValidatorManager::record_collate_query_stats, block_candidate->id,
+                          std::move(stats_));
   return true;
 }
 

validator/impl/validate-query.cpp

@@ -65,7 +65,7 @@ std::string ErrorCtx::as_string() const {
  * @param manager The ActorId of the ValidatorManager.
  * @param timeout The timeout for the validation.
  * @param promise The Promise to return the ValidateCandidateResult to.
- * @param is_fake A boolean indicating if the validation is fake (performed when creating a hardfork).
+ * @param mode +1 - fake mode
  */
 ValidateQuery::ValidateQuery(ShardIdFull shard, BlockIdExt min_masterchain_block_id,
                              std::vector<BlockIdExt> prev, BlockCandidate candidate, Ref<ValidatorSet> validator_set,

validator/interfaces/validator-manager.h

@@ -54,13 +54,24 @@ struct AsyncSerializerState {
 };
 
 struct CollationStats {
-  td::uint32 bytes, gas, lt_delta;
+  td::uint32 actual_bytes = 0, actual_collated_data_bytes = 0;
-  int cat_bytes, cat_gas, cat_lt_delta;
+  td::uint32 estimated_bytes = 0, gas = 0, lt_delta = 0, estimated_collated_data_bytes = 0;
+  int cat_bytes = 0, cat_gas = 0, cat_lt_delta = 0, cat_collated_data_bytes = 0;
   std::string limits_log;
   td::uint32 ext_msgs_total = 0;
   td::uint32 ext_msgs_filtered = 0;
   td::uint32 ext_msgs_accepted = 0;
   td::uint32 ext_msgs_rejected = 0;
+  double work_time = 0.0, cpu_work_time = 0.0;
+  td::uint32 serialized_size = 0, serialized_size_no_collated_data = 0;
+
+  tl_object_ptr<ton_api::validatorSession_collationStats> tl() const {
+    return create_tl_object<ton_api::validatorSession_collationStats>(
+        actual_bytes, actual_collated_data_bytes, estimated_bytes, gas, lt_delta, estimated_collated_data_bytes,
+        cat_bytes, cat_gas, cat_lt_delta, cat_collated_data_bytes, limits_log, ext_msgs_total, ext_msgs_filtered,
+        ext_msgs_accepted, ext_msgs_rejected, work_time, cpu_work_time, serialized_size,
+        serialized_size_no_collated_data);
+  }
 };
 
 using ValidateCandidateResult = td::Variant<UnixTime, CandidateReject>;
@@ -208,8 +219,7 @@ class ValidatorManager : public ValidatorManagerInterface {
   virtual void add_lite_query_stats(int lite_query_id) {
   }
 
-  virtual void record_collate_query_stats(BlockIdExt block_id, double work_time, double cpu_work_time,
+  virtual void record_collate_query_stats(BlockIdExt block_id, CollationStats stats) {
-                                          CollationStats stats) {
   }
   virtual void record_validate_query_stats(BlockIdExt block_id, double work_time, double cpu_work_time) {
   }

validator/manager.cpp

@@ -3099,10 +3099,7 @@ void ValidatorManagerImpl::log_validator_session_stats(BlockIdExt block_id,
       tl_object_ptr<ton_api::validatorSession_collationStats> collation_stats;
       if (it != recorded_block_stats_.end() && it->second.collator_stats_) {
         auto &stats = it->second.collator_stats_.value();
-        collation_stats = create_tl_object<ton_api::validatorSession_collationStats>(
+        collation_stats = stats.tl();
-            stats.bytes, stats.gas, stats.lt_delta, stats.cat_bytes, stats.cat_gas, stats.cat_lt_delta,
-            stats.limits_log, stats.ext_msgs_total, stats.ext_msgs_filtered, stats.ext_msgs_accepted,
-            stats.ext_msgs_rejected);
       }
       std::string approvers, signers;
       for (bool x : producer.approvers) {
@@ -3616,12 +3613,28 @@ td::actor::ActorOwn<ValidatorManagerInterface> ValidatorManagerFactory::create(
                                                                   rldp, overlays);
 }
 
-void ValidatorManagerImpl::record_collate_query_stats(BlockIdExt block_id, double work_time, double cpu_work_time,
+void ValidatorManagerImpl::record_collate_query_stats(BlockIdExt block_id, CollationStats stats) {
-                                                      CollationStats stats) {
   auto &record = new_block_stats_record(block_id);
-  record.collator_work_time_ = work_time;
+  record.collator_work_time_ = stats.work_time;
-  record.collator_cpu_work_time_ = cpu_work_time;
+  record.collator_cpu_work_time_ = stats.cpu_work_time;
   record.collator_stats_ = std::move(stats);
+
+  std::string fname = opts_->get_session_logs_file();
+  if (fname.empty()) {
+    return;
+  }
+
+  auto obj = create_tl_object<ton_api::validatorSession_statsCollatedBlock>(td::Clocks::system(),
+                                                                            create_tl_block_id(block_id), stats.tl());
+  auto s = td::json_encode<std::string>(td::ToJson(*obj.get()), false);
+  s.erase(std::remove_if(s.begin(), s.end(), [](char c) { return c == '\n' || c == '\r'; }), s.end());
+
+  std::ofstream file;
+  file.open(fname, std::ios_base::app);
+  file << s << "\n";
+  file.close();
+
+  LOG(DEBUG) << "Writing collation stats stats for " << block_id.id.to_str();
 }
 
 void ValidatorManagerImpl::record_validate_query_stats(BlockIdExt block_id, double work_time, double cpu_work_time) {

validator/manager.hpp

@@ -780,8 +780,7 @@ class ValidatorManagerImpl : public ValidatorManager {
   std::map<BlockIdExt, RecordedBlockStats> recorded_block_stats_;
   std::queue<BlockIdExt> recorded_block_stats_lru_;
 
-  void record_collate_query_stats(BlockIdExt block_id, double work_time, double cpu_work_time,
+  void record_collate_query_stats(BlockIdExt block_id, CollationStats stats) override;
-                                  CollationStats stats) override;
   void record_validate_query_stats(BlockIdExt block_id, double work_time, double cpu_work_time) override;
   RecordedBlockStats &new_block_stats_record(BlockIdExt block_id);
 

validator/validator.h

@@ -70,6 +70,11 @@ struct CollatorOptions : public td::CntObject {
   std::set<std::pair<WorkchainId, StdSmcAddress>> whitelist;
   // Prioritize these accounts on each phase of process_dispatch_queue
   std::set<std::pair<WorkchainId, StdSmcAddress>> prioritylist;
+
+  // Always enable full_collated_data
+  bool force_full_collated_data = false;
+  // Ignore collated data size limits from block limits and catchain config
+  bool ignore_collated_data_limits = false;
 };
 
 struct CollatorsList : public td::CntObject {