@@ -60,7 +60,7 @@ namespace {
6060 case SM::Fast:
6161 case SM::FastWithoutState:
6262 return kagome::network::BlockAttribute::HEADER
63- | kagome::network::BlockAttribute::JUSTIFICATION;
63+ | kagome::network::BlockAttribute::JUSTIFICATION;
6464 }
6565 return kagome::network::BlocksRequest::kBasicAttributes ;
6666 }
@@ -203,7 +203,9 @@ namespace kagome::network {
203203 it != known_blocks_.end ()) {
204204 auto &block_in_queue = it->second ;
205205 block_in_queue.peers .emplace (peer_id);
206- if (handler) handler (block_info);
206+ if (handler) {
207+ handler (block_info);
208+ }
207209 return false ;
208210 }
209211
@@ -229,7 +231,9 @@ namespace kagome::network {
229231
230232 // Provided block is equal our best one. Nothing needs to do.
231233 if (block_info == best_block) {
232- if (handler) handler (block_info);
234+ if (handler) {
235+ handler (block_info);
236+ }
233237 return false ;
234238 }
235239
@@ -264,7 +268,9 @@ namespace kagome::network {
264268
265269 // Finding the best common block was failed
266270 if (not res.has_value ()) {
267- if (handler) handler (res.as_failure ());
271+ if (handler) {
272+ handler (res.as_failure ());
273+ }
268274 return ;
269275 }
270276
@@ -274,7 +280,9 @@ namespace kagome::network {
274280 it != self->known_blocks_ .end ()) {
275281 auto &block_in_queue = it->second ;
276282 block_in_queue.peers .emplace (peer_id);
277- if (handler) handler (std::move (block_info));
283+ if (handler) {
284+ handler (std::move (block_info));
285+ }
278286 return ;
279287 }
280288
@@ -405,14 +413,16 @@ namespace kagome::network {
405413
406414 auto request_fingerprint = request.fingerprint ();
407415
408- if (not recent_requests_.emplace (peer_id, request_fingerprint).second ) {
416+ if (auto r = recent_requests_.emplace (
417+ std::make_tuple (peer_id, request_fingerprint), " find common block"
418+ not r.second ) {
409419 SL_VERBOSE (log_,
410420 " Can't check if block #{} in #{}..#{} is common with {}: {}"
411421 hint,
412422 lower,
413423 upper - 1 ,
414424 peer_id,
415- make_error_code (Error::DUPLICATE_REQUEST) );
425+ r. first -> second );
416426 handler (Error::DUPLICATE_REQUEST);
417427 return ;
418428 }
@@ -566,7 +576,9 @@ namespace kagome::network {
566576 SyncResultHandler &&handler) {
567577 // Interrupts process if node is shutting down
568578 if (node_is_shutting_down_) {
569- if (handler) handler (Error::SHUTTING_DOWN);
579+ if (handler) {
580+ handler (Error::SHUTTING_DOWN);
581+ }
570582 return ;
571583 }
572584
@@ -577,13 +589,17 @@ namespace kagome::network {
577589
578590 auto request_fingerprint = request.fingerprint ();
579591
580- if (not recent_requests_.emplace (peer_id, request_fingerprint).second ) {
592+ if (auto r = recent_requests_.emplace (
593+ std::make_tuple (peer_id, request_fingerprint), " load blocks"
594+ not r.second ) {
581595 SL_ERROR (log_,
582596 " Can't load blocks from {} beginning block {}: {}"
583597 peer_id,
584598 from,
585- make_error_code (Error::DUPLICATE_REQUEST));
586- if (handler) handler (Error::DUPLICATE_REQUEST);
599+ r.first ->second );
600+ if (handler) {
601+ handler (Error::DUPLICATE_REQUEST);
602+ }
587603 return ;
588604 }
589605
@@ -607,7 +623,9 @@ namespace kagome::network {
607623 peer_id,
608624 from,
609625 response_res.error ());
610- if (handler) handler (response_res.as_failure ());
626+ if (handler) {
627+ handler (response_res.as_failure ());
628+ }
611629 return ;
612630 }
613631 auto &blocks = response_res.value ().blocks ;
@@ -620,7 +638,9 @@ namespace kagome::network {
620638 " Response does not have any blocks"
621639 peer_id,
622640 from);
623- if (handler) handler (Error::EMPTY_RESPONSE);
641+ if (handler) {
642+ handler (Error::EMPTY_RESPONSE);
643+ }
624644 return ;
625645 }
626646
@@ -641,7 +661,9 @@ namespace kagome::network {
641661 " Received block without header"
642662 peer_id,
643663 from);
644- if (handler) handler (Error::RESPONSE_WITHOUT_BLOCK_HEADER);
664+ if (handler) {
665+ handler (Error::RESPONSE_WITHOUT_BLOCK_HEADER);
666+ }
645667 return ;
646668 }
647669 // Check if body is provided
@@ -651,7 +673,9 @@ namespace kagome::network {
651673 " Received block without body"
652674 peer_id,
653675 from);
654- if (handler) handler (Error::RESPONSE_WITHOUT_BLOCK_BODY);
676+ if (handler) {
677+ handler (Error::RESPONSE_WITHOUT_BLOCK_BODY);
678+ }
655679 return ;
656680 }
657681 auto &header = block.header .value ();
@@ -669,7 +693,9 @@ namespace kagome::network {
669693 peer_id,
670694 from,
671695 BlockInfo (header.number , block.hash ));
672- if (handler) handler (Error::DISCARDED_BLOCK);
696+ if (handler) {
697+ handler (Error::DISCARDED_BLOCK);
698+ }
673699 return ;
674700 }
675701
@@ -700,7 +726,9 @@ namespace kagome::network {
700726 peer_id,
701727 from,
702728 BlockInfo (header.number , header.parent_hash ));
703- if (handler) handler (Error::DISCARDED_BLOCK);
729+ if (handler) {
730+ handler (Error::DISCARDED_BLOCK);
731+ }
704732 return ;
705733 }
706734
@@ -716,7 +744,9 @@ namespace kagome::network {
716744 " block {}: Received block is not descendant of previous"
717745 peer_id,
718746 from);
719- if (handler) handler (Error::WRONG_ORDER);
747+ if (handler) {
748+ handler (Error::WRONG_ORDER);
749+ }
720750 return ;
721751 }
722752
@@ -730,7 +760,9 @@ namespace kagome::network {
730760 " Received block whose hash does not match the header"
731761 peer_id,
732762 from);
733- if (handler) handler (Error::INVALID_HASH);
763+ if (handler) {
764+ handler (Error::INVALID_HASH);
765+ }
734766 return ;
735767 }
736768
@@ -788,7 +820,9 @@ namespace kagome::network {
788820 std::optional<uint32_t > limit,
789821 SyncResultHandler &&handler) {
790822 if (node_is_shutting_down_) {
791- if (handler) handler (Error::SHUTTING_DOWN);
823+ if (handler) {
824+ handler (Error::SHUTTING_DOWN);
825+ }
792826 return ;
793827 }
794828
@@ -807,12 +841,15 @@ namespace kagome::network {
807841 limit};
808842
809843 auto request_fingerprint = request.fingerprint ();
810- if (not recent_requests_.emplace (peer_id, request_fingerprint).second ) {
844+ if (auto r = recent_requests_.emplace (
845+ std::make_tuple (peer_id, request_fingerprint),
846+ " load justifications"
847+ not r.second ) {
811848 SL_ERROR (log_,
812849 " Can't load justification from {} for block {}: {}"
813850 peer_id,
814851 target_block,
815- make_error_code (Error::DUPLICATE_REQUEST) );
852+ r. first -> second );
816853 if (handler) {
817854 handler (Error::DUPLICATE_REQUEST);
818855 }
@@ -824,6 +861,7 @@ namespace kagome::network {
824861 auto response_handler = [wp = weak_from_this (),
825862 peer_id,
826863 target_block,
864+ limit,
827865 handler = std::move (handler)](
828866 auto &&response_res) mutable {
829867 auto self = wp.lock ();
@@ -851,26 +889,42 @@ namespace kagome::network {
851889 " Response does not have any contents"
852890 peer_id,
853891 target_block);
854- if (handler) handler (Error::EMPTY_RESPONSE);
892+ if (handler) {
893+ handler (Error::EMPTY_RESPONSE);
894+ }
855895 return ;
856896 }
857897
898+ // Use decreasing limit,
899+ // to avoid race between block and justification requests
900+ if (limit.has_value ()) {
901+ if (blocks.size () >= limit.value ()) {
902+ limit = 0 ;
903+ } else {
904+ limit.value () -= (blocks.size () - 1 );
905+ }
906+ }
907+
858908 bool justification_received = false ;
859909 BlockInfo last_justified_block;
910+ BlockInfo last_observed_block;
860911 for (auto &block : blocks) {
861912 if (not block.header ) {
862913 SL_ERROR (self->log_ ,
863914 " No header was provided from {} for block {} while "
864915 " requesting justifications"
865916 peer_id,
866917 target_block);
867- if (handler) handler (Error::RESPONSE_WITHOUT_BLOCK_HEADER);
918+ if (handler) {
919+ handler (Error::RESPONSE_WITHOUT_BLOCK_HEADER);
920+ }
868921 return ;
869922 }
923+ last_observed_block =
924+ primitives::BlockInfo{block.header ->number , block.hash };
870925 if (block.justification ) {
871926 justification_received = true ;
872- last_justified_block =
873- primitives::BlockInfo{block.header ->number , block.hash };
927+ last_justified_block = last_observed_block;
874928 {
875929 std::lock_guard lock (self->justifications_mutex_ );
876930 self->justifications_ .emplace (last_justified_block,
@@ -887,6 +941,25 @@ namespace kagome::network {
887941 }
888942 });
889943 }
944+
945+ // Continue justifications requesting till limit is non-zero and last
946+ // observed block is not target (no block anymore)
947+ if ((not limit.has_value () or limit.value () > 0 )
948+ and last_observed_block != target_block) {
949+ SL_TRACE (self->log_ , " Request next block pack"
950+ self->scheduler_ ->schedule ([wp,
951+ peer_id,
952+ target_block = last_observed_block,
953+ limit,
954+ handler = std::move (handler)]() mutable {
955+ if (auto self = wp.lock ()) {
956+ self->loadJustifications (
957+ peer_id, target_block, limit, std::move (handler));
958+ }
959+ });
960+ return ;
961+ }
962+
890963 if (handler) {
891964 handler (last_justified_block);
892965 }
@@ -1043,7 +1116,9 @@ namespace kagome::network {
10431116 " Block {} {} not applied as discarded"
10441117 block_info,
10451118 n ? fmt::format (" and {} others have" fmt::format (" has"
1046- if (handler) handler (Error::DISCARDED_BLOCK);
1119+ if (handler) {
1120+ handler (Error::DISCARDED_BLOCK);
1121+ }
10471122 }
10481123
10491124 } else {
@@ -1068,7 +1143,9 @@ namespace kagome::network {
10681143 " state syncing on block in progress"
10691144 block_info,
10701145 n ? fmt::format (" and {} others have" fmt::format (" has"
1071- if (handler) handler (Error::DISCARDED_BLOCK);
1146+ if (handler) {
1147+ handler (Error::DISCARDED_BLOCK);
1148+ }
10721149 return ;
10731150 }
10741151 }
@@ -1086,15 +1163,21 @@ namespace kagome::network {
10861163 block_info,
10871164 n ? fmt::format (" and {} others have" fmt::format (" has"
10881165 applying_res.error ());
1089- if (handler) handler (Error::DISCARDED_BLOCK);
1166+ if (handler) {
1167+ handler (Error::DISCARDED_BLOCK);
1168+ }
10901169 } else {
10911170 SL_DEBUG (log_, " Block {} is skipped as existing"
1092- if (handler) handler (block_info);
1171+ if (handler) {
1172+ handler (block_info);
1173+ }
10931174 }
10941175 } else {
10951176 telemetry_->notifyBlockImported (
10961177 block_info, telemetry::BlockOrigin::kNetworkInitialSync );
1097- if (handler) handler (block_info);
1178+ if (handler) {
1179+ handler (block_info);
1180+ }
10981181
10991182 // Check if finality lag greater than justification saving interval
11001183 static const BlockNumber kJustificationInterval = 512 ;
@@ -1108,7 +1191,7 @@ namespace kagome::network {
11081191 syncMissingJustifications (
11091192 peer_id,
11101193 last_finalized,
1111- std::nullopt ,
1194+ kJustificationInterval * 2 ,
11121195 [wp = weak_from_this (), last_finalized, block_info](
11131196 auto res) {
11141197 if (auto self = wp.lock ()) {
0 commit comments