Collator: change deferring behavior when out queue size is big (#1135)

tl/generate/scheme/ton_api.tl

@@ -628,7 +628,8 @@ engine.validator.customOverlaysConfig overlays:(vector engine.validator.customOv
 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 = engine.validator.CollatorOptions;
+  dispatch_phase_2_max_per_initiator:int dispatch_phase_3_max_per_initiator:int
+  whitelist:(vector string) prioritylist:(vector string) = engine.validator.CollatorOptions;
 
 ---functions---
 ---types---

validator-engine/validator-engine.cpp

@@ -2531,6 +2531,14 @@ static td::Result<td::Ref<ton::validator::CollatorOptions>> parse_collator_optio
   } else {
     opts.dispatch_phase_3_max_per_initiator = {};
   }
+  for (const std::string& s : f.whitelist_) {
+    TRY_RESULT(addr, block::StdAddress::parse(s));
+    opts.whitelist.emplace(addr.workchain, addr.addr);
+  }
+  for (const std::string& s : f.prioritylist_) {
+    TRY_RESULT(addr, block::StdAddress::parse(s));
+    opts.prioritylist.emplace(addr.workchain, addr.addr);
+  }
 
   return ref;
 }

validator/impl/collator.cpp

@@ -3066,7 +3066,7 @@ int Collator::process_one_new_message(block::NewOutMsg msg, bool enqueue_only, R
   bool defer = false;
   if (!from_dispatch_queue) {
     if (deferring_messages_enabled_ && collator_opts_->deferring_enabled && !is_special && !is_special_account &&
-        msg.msg_idx != 0) {
+        !collator_opts_->whitelist.count({src_wc, src_addr}) && msg.msg_idx != 0) {
       if (++sender_generated_messages_count_[src_addr] >= collator_opts_->defer_messages_after ||
           out_msg_queue_size_ > defer_out_queue_size_limit_) {
         defer = true;
@@ -3697,6 +3697,8 @@ bool Collator::process_dispatch_queue() {
     vm::AugmentedDictionary cur_dispatch_queue{dispatch_queue_->get_root(), 256, block::tlb::aug_DispatchQueue};
     std::map<std::tuple<WorkchainId, StdSmcAddress, LogicalTime>, size_t> count_per_initiator;
     size_t total_count = 0;
+    auto prioritylist = collator_opts_->prioritylist;
+    auto prioritylist_iter = prioritylist.begin();
     while (!cur_dispatch_queue.is_empty()) {
       block_full_ = !block_limit_status_->fits(block::ParamLimits::cl_normal);
       if (block_full_) {
@@ -3713,9 +3715,30 @@ bool Collator::process_dispatch_queue() {
         return true;
       }
       StdSmcAddress src_addr;
-      auto account_dispatch_queue = block::get_dispatch_queue_min_lt_account(cur_dispatch_queue, src_addr);
+      td::Ref<vm::CellSlice> account_dispatch_queue;
+      while (!prioritylist.empty()) {
+        if (prioritylist_iter == prioritylist.end()) {
+          prioritylist_iter = prioritylist.begin();
+        }
+        auto priority_addr = *prioritylist_iter;
+        if (priority_addr.first != workchain() || !is_our_address(priority_addr.second)) {
+          prioritylist_iter = prioritylist.erase(prioritylist_iter);
+          continue;
+        }
+        src_addr = priority_addr.second;
+        account_dispatch_queue = cur_dispatch_queue.lookup(src_addr);
+        if (account_dispatch_queue.is_null()) {
+          prioritylist_iter = prioritylist.erase(prioritylist_iter);
+        } else {
+          ++prioritylist_iter;
+          break;
+        }
+      }
       if (account_dispatch_queue.is_null()) {
-        return fatal_error("invalid dispatch queue in shard state");
+        account_dispatch_queue = block::get_dispatch_queue_min_lt_account(cur_dispatch_queue, src_addr);
+        if (account_dispatch_queue.is_null()) {
+          return fatal_error("invalid dispatch queue in shard state");
+        }
       }
       vm::Dictionary dict{64};
       td::uint64 dict_size;
@@ -3735,7 +3758,8 @@ bool Collator::process_dispatch_queue() {
       // Remove message from DispatchQueue
       bool ok;
       if (iter == 0 ||
-          (iter == 1 && sender_generated_messages_count_[src_addr] >= collator_opts_->defer_messages_after)) {
+          (iter == 1 && sender_generated_messages_count_[src_addr] >= collator_opts_->defer_messages_after &&
+           !collator_opts_->whitelist.count({workchain(), src_addr}))) {
         ok = cur_dispatch_queue.lookup_delete(src_addr).not_null();
       } else {
         dict.lookup_delete(key);

validator/validator.h

@@ -64,6 +64,11 @@ struct CollatorOptions : public td::CntObject {
   td::uint32 dispatch_phase_3_max_total = 150;
   td::uint32 dispatch_phase_2_max_per_initiator = 20;
   td::optional<td::uint32> dispatch_phase_3_max_per_initiator;  // Default - depends on out msg queue size
+
+  // Don't defer messages from these accounts
+  std::set<std::pair<WorkchainId, StdSmcAddress>> whitelist;
+  // Prioritize these accounts on each phase of process_dispatch_queue
+  std::set<std::pair<WorkchainId, StdSmcAddress>> prioritylist;
 };
 
 struct ValidatorManagerOptions : public td::CntObject {