detect closed read streams (#2360)

Signed-off-by: turuslan <[email protected]>
qdrvm · Jan 29, 2025 · 352e38f · 352e38f
1 parent 97406d9
commit 352e38f
Show file tree

Hide file tree

Showing 2 changed files with 39 additions and 8 deletions.
diff --git a/core/network/impl/synchronizer_impl.cpp b/core/network/impl/synchronizer_impl.cpp
@@ -1476,6 +1476,15 @@ namespace kagome::network {
   }
 
   void SynchronizerImpl::randomWarp() {
+    scheduler_->schedule(
+        [WEAK_SELF] {
+          WEAK_LOCK(self);
+          self->randomWarp();
+        },
+        kRandomWarpInterval);
+    if (not timeline_.get()->wasSynchronized()) {
+      return;
+    }
     auto finalized = block_tree_->getLastFinalized();
     auto cb = [WEAK_SELF, finalized](outcome::result<WarpResponse> r) mutable {
       WEAK_LOCK(self);
@@ -1495,11 +1504,5 @@ namespace kagome::network {
       }
     };
     router_->getWarpProtocol()->random(finalized.hash, cb);
-    scheduler_->schedule(
-        [WEAK_SELF] {
-          WEAK_LOCK(self);
-          self->randomWarp();
-        },
-        kRandomWarpInterval);
   }
 }  // namespace kagome::network
diff --git a/core/network/notifications/protocol.cpp b/core/network/notifications/protocol.cpp
@@ -20,6 +20,11 @@ namespace kagome::network::notifications {
   constexpr auto kBackoffMin = std::chrono::seconds{5};
   constexpr auto kBackoffMax = std::chrono::seconds{10};
 
+  // TODO(turuslan): #2359, remove when `YamuxStream::readSome` returns error
+  inline bool isClosed(const StreamInfoClose &stream) {
+    return stream.stream->isClosed();
+  }
+
   StreamInfo::StreamInfo(const ProtocolsGroups &protocols_groups,
                          const StreamAndProtocol &info)
       : protocol_group{},
@@ -331,6 +336,10 @@ namespace kagome::network::notifications {
     if (not stream) {
       return;
     }
+    if (isClosed(*stream)) {
+      onError(peer_id, false);
+      return;
+    }
     auto cb = [WEAK_SELF, peer_id, protocol_group{stream->protocol_group}](
                   libp2p::basic::MessageReadWriter::ReadCallback r) mutable {
       WEAK_LOCK(self);
@@ -376,6 +385,14 @@ namespace kagome::network::notifications {
     if (controller_.expired()) {
       return;
     }
+    for (auto it = peers_in_.begin(); it != peers_in_.end();) {
+      auto &[peer_id, stream] = *it;
+      ++it;
+      if (isClosed(stream)) {
+        // copy `it->first` before `erase(it)`
+        onError(PeerId{peer_id}, false);
+      }
+    }
     for (auto &peer_id : reserved_) {
       open(peer_id);
     }
@@ -407,6 +424,9 @@ namespace kagome::network::notifications {
     size_t count = 0;
     if (not out) {
       for (auto &p : peers_in_) {
+        if (isClosed(p.second)) {
+          continue;
+        }
         if (reserved_.contains(p.first)) {
           continue;
         }
@@ -427,8 +447,16 @@ namespace kagome::network::notifications {
   }
 
   bool Protocol::shouldAccept(const PeerId &peer_id) {
-    if (peers_in_.contains(peer_id)) {
-      return false;
+    auto peer_in = entry(peers_in_, peer_id);
+    if (peer_in) {
+      // if stream was closed, but read didn't return error
+      if (isClosed(*peer_in)) {
+        // remove closed stream
+        onError(peer_id, false);
+        // no `return`, continue `shouldAccept` checks
+      } else {
+        return false;
+      }
     }
     if (reserved_.contains(peer_id)) {
       return true;