diff --git a/cmake/Hunter/config.cmake b/cmake/Hunter/config.cmake
index 60760abab..1ffd537ba 100644
--- a/cmake/Hunter/config.cmake
+++ b/cmake/Hunter/config.cmake
@@ -15,27 +15,9 @@
 #     CMAKE_ARGS "CMAKE_VARIABLE=value"
 # )
 
-hunter_config(
-    soralog
-    VERSION 0.2.5
-    URL  https://github.com/qdrvm/soralog/archive/refs/tags/v0.2.5.tar.gz
-    SHA1 1dafdb9e1921b4069f9e1dad0d0acfae24166bd2
-    KEEP_PACKAGE_SOURCES
-)
-
 hunter_config(
     ZLIB
     VERSION 1.3.0-p0
     URL  https://github.com/cpp-pm/zlib/archive/refs/tags/v1.3.0-p0.tar.gz
     SHA1 311ca59e20cbbfe9d9e05196c12c6ae109093987
 )
-
-hunter_config(
-    qtils
-    VERSION 0.1.0
-    URL  https://github.com/qdrvm/qtils/archive/refs/tags/v0.1.0.tar.gz
-    SHA1 acc28902af7dc5d74ac33d486ad2261906716f5e
-    CMAKE_ARGS
-      FORMAT_ERROR_WITH_FULLTYPE=ON
-    KEEP_PACKAGE_SOURCES
-)
diff --git a/cmake/Hunter/init.cmake b/cmake/Hunter/init.cmake
index 6f8fc7f35..3e6981fab 100644
--- a/cmake/Hunter/init.cmake
+++ b/cmake/Hunter/init.cmake
@@ -31,7 +31,7 @@ set(
 include(${CMAKE_CURRENT_LIST_DIR}/HunterGate.cmake)
 
 HunterGate(
-    URL  https://github.com/qdrvm/hunter/archive/refs/tags/v0.25.3-qdrvm28.tar.gz
-    SHA1 a4f1b0f42464e07790b7f90b783a822d71be6c6d
+    URL  https://github.com/qdrvm/hunter/archive/86e53c752977bf5a5efb6590c26941ed3fec8187.zip
+    SHA1 7121f2cbf052cc6cb0a526a89e9b4aca5bf3cd54
     LOCAL
 )
diff --git a/include/libp2p/basic/scheduler/manual_scheduler_backend.hpp b/include/libp2p/basic/scheduler/manual_scheduler_backend.hpp
index 1a1ece46e..8fe9deff9 100644
--- a/include/libp2p/basic/scheduler/manual_scheduler_backend.hpp
+++ b/include/libp2p/basic/scheduler/manual_scheduler_backend.hpp
@@ -66,9 +66,9 @@ namespace libp2p::basic {
       }
     }
 
-   private:
     void callDeferred();
 
+   private:
     /// Current time, set manually
     std::chrono::milliseconds current_clock_;
 
diff --git a/include/libp2p/connection/stream_pair.hpp b/include/libp2p/connection/stream_pair.hpp
new file mode 100644
index 000000000..74104abb0
--- /dev/null
+++ b/include/libp2p/connection/stream_pair.hpp
@@ -0,0 +1,24 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <libp2p/peer/peer_id.hpp>
+
+namespace libp2p::basic {
+  class Scheduler;
+}  // namespace libp2p::basic
+
+namespace libp2p::connection {
+  struct Stream;
+
+  /**
+   * Create pair of connected bidirectional read-writers
+   * implementing `Stream` interface.
+   */
+  std::pair<std::shared_ptr<Stream>, std::shared_ptr<Stream>> streamPair(
+      std::shared_ptr<basic::Scheduler> post, PeerId peer1, PeerId peer2);
+}  // namespace libp2p::connection
diff --git a/include/libp2p/peer/protocol.hpp b/include/libp2p/peer/protocol.hpp
index 67fae1f25..b8b87561f 100644
--- a/include/libp2p/peer/protocol.hpp
+++ b/include/libp2p/peer/protocol.hpp
@@ -17,3 +17,7 @@ namespace libp2p::peer {
       std::string;
 
 }  // namespace libp2p::peer
+
+namespace libp2p {
+  using peer::ProtocolName;
+}  // namespace libp2p
diff --git a/include/libp2p/protocol/gossip/explicit_peers.hpp b/include/libp2p/protocol/gossip/explicit_peers.hpp
new file mode 100644
index 000000000..43212ae04
--- /dev/null
+++ b/include/libp2p/protocol/gossip/explicit_peers.hpp
@@ -0,0 +1,18 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <libp2p/peer/peer_id.hpp>
+
+namespace libp2p::protocol::gossip {
+  class ExplicitPeers {
+   public:
+    bool contains(const PeerId &) const {
+      return false;
+    }
+  };
+}  // namespace libp2p::protocol::gossip
diff --git a/include/libp2p/protocol/gossip/gossip.hpp b/include/libp2p/protocol/gossip/gossip.hpp
index 0b289db4d..d98328d97 100644
--- a/include/libp2p/protocol/gossip/gossip.hpp
+++ b/include/libp2p/protocol/gossip/gossip.hpp
@@ -10,6 +10,7 @@
 #include <functional>
 #include <set>
 #include <string>
+#include <unordered_map>
 #include <vector>
 
 #include <boost/optional.hpp>
@@ -17,7 +18,10 @@
 #include <libp2p/common/byteutil.hpp>
 #include <libp2p/multi/multiaddress.hpp>
 #include <libp2p/peer/peer_id.hpp>
+#include <libp2p/peer/protocol.hpp>
 #include <libp2p/protocol/common/subscription.hpp>
+#include <libp2p/protocol/gossip/peer_kind.hpp>
+#include <libp2p/protocol/gossip/score_config.hpp>
 
 namespace libp2p {
   struct Host;
@@ -41,8 +45,24 @@ namespace libp2p::protocol::gossip {
   struct Config {
     /// Network density factors for gossip meshes
     size_t D_min = 5;
+    size_t D_lazy = 6;
+    size_t D = 6;
     size_t D_max = 10;
 
+    /// Affects how peers are selected when pruning a mesh due to over
+    /// subscription.
+    ///
+    ///  At least `retain_scores` of the retained peers will be high-scoring,
+    ///  while the remainder
+    /// are  chosen randomly (D_score in the spec, default is 4).
+    size_t retain_scores = 4;
+
+    /// Affects how many peers we will emit gossip to at each heartbeat.
+    ///
+    /// We will send gossip to `gossip_factor * (total number of non-mesh
+    /// peers)`, or `gossip_lazy`, whichever is greater. The default is 0.25.
+    double gossip_factor = 0.25;
+
     /// Ideal number of connected peers to support the network
     size_t ideal_connections_num = 100;
 
@@ -50,27 +70,12 @@ namespace libp2p::protocol::gossip {
     /// incoming peers will be rejected
     size_t max_connections_num = 1000;
 
-    /// Forward messages to all subscribers not in mesh
-    /// (floodsub mode compatibility)
-    bool floodsub_forward_mode = false;
-
     /// Forward local message to local subscribers
     bool echo_forward_mode = false;
 
     /// Read or write timeout per whole network operation
     std::chrono::milliseconds rw_timeout_msec{std::chrono::seconds(10)};
 
-    /// Lifetime of a message in message cache
-    std::chrono::milliseconds message_cache_lifetime_msec{
-        std::chrono::minutes(2)};
-
-    /// Topic's message seen cache lifetime
-    std::chrono::milliseconds seen_cache_lifetime_msec{
-        message_cache_lifetime_msec * 3 / 4};
-
-    /// Topic's seen cache limit
-    unsigned seen_cache_limit = 100;
-
     /// Heartbeat interval
     std::chrono::milliseconds heartbeat_interval_msec{1000};
 
@@ -86,11 +91,117 @@ namespace libp2p::protocol::gossip {
     /// Max RPC message size
     size_t max_message_size = 1 << 24;
 
-    /// Protocol version
-    std::string protocol_version = "/meshsub/1.0.0";
+    /// Protocol versions
+    std::unordered_map<ProtocolName, PeerKind> protocol_versions{
+        {"/floodsub/1.0.0", PeerKind::Floodsub},
+        {"/meshsub/1.0.0", PeerKind::Gossipsub},
+        {"/meshsub/1.1.0", PeerKind::Gossipsubv1_1},
+        {"/meshsub/1.2.0", PeerKind::Gossipsubv1_2},
+    };
 
     /// Sign published messages
     bool sign_messages = false;
+
+    /// Number of heartbeats to keep in the `memcache`
+    size_t history_length{5};
+
+    /// Number of past heartbeats to gossip about (default is 3).
+    size_t history_gossip{3};
+
+    /// Time to live for fanout peers (default is 60 seconds).
+    std::chrono::seconds fanout_ttl{60};
+
+    /// Duplicates are prevented by storing message id's of known messages in an
+    /// LRU time cache. This settings sets the time period that messages are
+    /// stored in the cache. Duplicates can be received if duplicate messages
+    /// are sent at a time greater than this setting apart. The default is 1
+    /// minute.
+    std::chrono::seconds duplicate_cache_time{60};
+
+    /// Controls the backoff time for pruned peers. This is how long
+    /// a peer must wait before attempting to graft into our mesh again after
+    /// being pruned. When pruning a peer, we send them our value of
+    /// `prune_backoff` so they know the minimum time to wait. Peers running
+    /// older versions may not send a backoff time, so if we receive a prune
+    /// message without one, we will wait at least `prune_backoff` before
+    /// attempting to re-graft. The default is one minute.
+    std::chrono::seconds prune_backoff{60};
+
+    /// Controls the backoff time when unsubscribing from a topic.
+    ///
+    /// This is how long to wait before resubscribing to the topic. A short
+    /// backoff period in case of an unsubscribe event allows reaching a healthy
+    /// mesh in a more timely manner. The default is 10 seconds.
+    std::chrono::seconds unsubscribe_backoff{10};
+
+    /// Number of heartbeat slots considered as slack for backoffs. This
+    /// guarantees that we wait at least backoff_slack heartbeats after a
+    /// backoff is over before we try to graft. This solves problems occurring
+    /// through high latencies. In particular if `backoff_slack *
+    /// heartbeat_interval` is longer than any latencies between processing
+    /// prunes on our side and processing prunes on the receiving side this
+    /// guarantees that we get not punished for too early grafting. The default
+    /// is 1.
+    size_t backoff_slack = 1;
+
+    /// Whether to do flood publishing or not. If enabled newly created messages
+    /// will always be
+    /// sent to all peers that are subscribed to the topic and have a good
+    /// enough score. The default is true.
+    bool flood_publish = true;
+
+    /// If a GRAFT comes before `graft_flood_threshold` has elapsed since the
+    /// last PRUNE, then there is an extra score penalty applied to the peer
+    /// through P7.
+    std::chrono::milliseconds graft_flood_threshold = std::chrono::seconds{10};
+
+    /// Minimum number of outbound peers in the mesh network before adding more
+    /// (D_out in the spec). This value must be smaller or equal than `mesh_n /
+    /// 2` and smaller than `mesh_n_low`. The default is 2.
+    size_t mesh_outbound_min = 2;
+
+    /// Number of heartbeat ticks that specify the interval in which
+    /// opportunistic grafting is applied. Every `opportunistic_graft_ticks` we
+    /// will attempt to select some high-scoring mesh peers to replace
+    /// lower-scoring ones, if the median score of our mesh peers falls below a
+    /// threshold (see
+    /// <https://godoc.org/github.com/libp2p/go-libp2p-pubsub#PeerScoreThresholds>).
+    /// The default is 60.
+    size_t opportunistic_graft_ticks = 60;
+
+    /// The maximum number of new peers to graft to during opportunistic
+    /// grafting. The default is 2.
+    size_t opportunistic_graft_peers = 2;
+
+    /// The maximum number of messages to include in an IHAVE message.
+    /// Also controls the maximum number of IHAVE ids we will accept and request
+    /// with IWANT from a peer within a heartbeat, to protect from IHAVE floods.
+    /// You should adjust this value from the default if your system is pushing
+    /// more than 5000 messages in GossipSubHistoryGossip heartbeats; with the
+    /// defaults this is 1666 messages/s. The default is 5000.
+    size_t max_ihave_length = 5000;
+
+    /// Time to wait for a message requested through IWANT following an IHAVE
+    /// advertisement. If the message is not received within this window, a
+    /// broken promise is declared and the router may apply behavioural
+    /// penalties. The default is 3 seconds.
+    std::chrono::milliseconds iwant_followup_time = std::chrono::seconds{3};
+
+    /// The message size threshold for which IDONTWANT messages are sent.
+    /// Sending IDONTWANT messages for small messages can have a negative effect
+    /// to the overall traffic and CPU load. This acts as a lower bound cutoff
+    /// for the message size to which IDONTWANT won't be sent to peers. Only
+    /// works if the peers support Gossipsub1.2 (see
+    /// <https://github.com/libp2p/specs/blob/master/pubsub/gossipsub/gossipsub-v1.2.md#idontwant-message>)
+    /// default is 1kB
+    size_t idontwant_message_size_threshold = 1000;
+
+    /// Send IDONTWANT messages after publishing message on gossip. This is an
+    /// optimisation to avoid bandwidth consumption by downloading the published
+    /// message over gossip. By default it is false.
+    bool idontwant_on_publish = false;
+
+    ScoreConfig score;
   };
 
   using TopicId = std::string;
diff --git a/include/libp2p/protocol/gossip/peer_kind.hpp b/include/libp2p/protocol/gossip/peer_kind.hpp
new file mode 100644
index 000000000..d0416f0de
--- /dev/null
+++ b/include/libp2p/protocol/gossip/peer_kind.hpp
@@ -0,0 +1,19 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <cstdint>
+
+namespace libp2p::protocol::gossip {
+  enum class PeerKind : uint8_t {
+    NotSupported,
+    Floodsub,
+    Gossipsub,
+    Gossipsubv1_1,
+    Gossipsubv1_2,
+  };
+}  // namespace libp2p::protocol::gossip
diff --git a/include/libp2p/protocol/gossip/score.hpp b/include/libp2p/protocol/gossip/score.hpp
new file mode 100644
index 000000000..ef699aa57
--- /dev/null
+++ b/include/libp2p/protocol/gossip/score.hpp
@@ -0,0 +1,340 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <chrono>
+#include <libp2p/peer/peer_id.hpp>
+#include <libp2p/protocol/gossip/score_config.hpp>
+#include <libp2p/protocol/gossip/time_cache.hpp>
+#include <unordered_map>
+#include <unordered_set>
+#include <variant>
+
+namespace libp2p::protocol::gossip {
+  using TopicId = std::string;
+  using MessageId = Bytes;
+}  // namespace libp2p::protocol::gossip
+
+namespace libp2p::protocol::gossip::score {
+  using Duration = std::chrono::milliseconds;
+  using Clock = std::chrono::steady_clock;
+  using Time = Clock::time_point;
+
+  constexpr std::chrono::seconds kTimeCacheDuration{120};
+
+  struct DeliveryStatusUnknown {};
+  struct DeliveryStatusValid {
+    Time time;
+  };
+  struct DeliveryStatusInvalid {};
+  struct DeliveryStatusIgnored {};
+  using DeliveryStatus = std::variant<DeliveryStatusUnknown,
+                                      DeliveryStatusValid,
+                                      DeliveryStatusInvalid,
+                                      DeliveryStatusIgnored>;
+
+  struct DeliveryRecord {
+    DeliveryStatus status;
+    Time first_seen;
+    std::unordered_set<PeerId> peers;
+  };
+
+  struct MeshActive {
+    Time graft_time;
+    Duration mesh_time;
+  };
+
+  struct TopicStats {
+    std::optional<MeshActive> mesh_active;
+    double first_message_deliveries = 0;
+    bool mesh_message_deliveries_active = false;
+    double mesh_message_deliveries = 0;
+    double mesh_failure_penalty = 0;
+    double invalid_message_deliveries = 0;
+  };
+
+  struct PeerStats {
+    std::optional<Time> expires_at;
+    std::unordered_map<TopicId, TopicStats> topics;
+    double behaviour_penalty = 0;
+    double application_score = 0;
+    double slow_peer_penalty = 0;
+  };
+
+  class Score {
+   public:
+    Score(ScoreConfig params)
+        : params_{std::move(params)}, deliveries_{kTimeCacheDuration} {}
+
+    bool below(const PeerId &peer_id, double threshold) {
+      return score(peer_id) < threshold;
+    }
+    double score(const PeerId &peer_id) {
+      auto it = peer_stats_.find(peer_id);
+      if (it == peer_stats_.end()) {
+        return 0;
+      }
+      auto &peer_stats = it->second;
+      double score = 0;
+      for (auto &[topic, topic_stats] : peer_stats.topics) {
+        auto topic_it = params_.topics.find(topic);
+        if (topic_it == params_.topics.end()) {
+          continue;
+        }
+        auto &topic_params = topic_it->second;
+        double topic_score = 0;
+        if (topic_stats.mesh_active.has_value()) {
+          auto as_secs_f64 = [](auto t) {
+            return static_cast<double>(
+                std::chrono::duration_cast<std::chrono::seconds>(t).count());
+          };
+          auto p1 =
+              std::min(as_secs_f64(topic_stats.mesh_active->mesh_time)
+                           / as_secs_f64(topic_params.time_in_mesh_quantum),
+                       topic_params.time_in_mesh_cap);
+          topic_score += p1 * topic_params.time_in_mesh_weight;
+        }
+        auto p2 = std::min(topic_stats.first_message_deliveries,
+                           topic_params.first_message_deliveries_cap);
+        topic_score += p2 * topic_params.first_message_deliveries_weight;
+        if (topic_stats.mesh_message_deliveries_active
+            and topic_stats.mesh_message_deliveries
+                    < topic_params.mesh_message_deliveries_threshold) {
+          auto deficit = topic_params.mesh_message_deliveries_threshold
+                       - topic_stats.mesh_message_deliveries;
+          auto p3 = deficit * deficit;
+          topic_score += p3 * topic_params.mesh_message_deliveries_weight;
+        }
+        auto p3b = topic_stats.mesh_failure_penalty;
+        topic_score += p3b * topic_params.mesh_failure_penalty_weight;
+        auto p4 = topic_stats.invalid_message_deliveries
+                * topic_stats.invalid_message_deliveries;
+        topic_score += p4 * topic_params.invalid_message_deliveries_weight;
+        score += topic_score * topic_params.topic_weight;
+      }
+      if (params_.topic_score_cap > 0 and score > params_.topic_score_cap) {
+        score = params_.topic_score_cap;
+      }
+      auto p5 = peer_stats.application_score;
+      score += p5 * params_.app_specific_weight;
+      if (peer_stats.behaviour_penalty > params_.behaviour_penalty_threshold) {
+        auto excess =
+            peer_stats.behaviour_penalty - params_.behaviour_penalty_threshold;
+        auto p7 = excess * excess;
+        score += p7 * params_.behaviour_penalty_weight;
+      }
+      if (peer_stats.slow_peer_penalty > params_.slow_peer_threshold) {
+        auto excess =
+            peer_stats.slow_peer_penalty - params_.slow_peer_threshold;
+        score += excess * params_.slow_peer_weight;
+      }
+      return score;
+    }
+    void addPenalty(const PeerId &peer_id, size_t count) {
+      auto it = peer_stats_.find(peer_id);
+      if (it == peer_stats_.end()) {
+        return;
+      }
+      it->second.behaviour_penalty += static_cast<double>(count);
+    }
+    void graft(const PeerId &peer_id, const TopicId &topic) {
+      auto it = peer_stats_.find(peer_id);
+      if (it == peer_stats_.end()) {
+        return;
+      }
+      if (auto *topic_stats = statsOrDefault(it->second, topic)) {
+        topic_stats->mesh_active = MeshActive{
+            .graft_time = Clock::now(),
+            .mesh_time = {},
+        };
+        topic_stats->mesh_message_deliveries_active = false;
+      }
+    }
+    void prune(const PeerId &peer_id, const TopicId &topic) {
+      auto it = peer_stats_.find(peer_id);
+      if (it == peer_stats_.end()) {
+        return;
+      }
+      if (auto *topic_stats = statsOrDefault(it->second, topic)) {
+        auto &threshold =
+            params_.topics.at(topic).mesh_message_deliveries_threshold;
+        if (topic_stats->mesh_message_deliveries_active
+            and topic_stats->mesh_message_deliveries < threshold) {
+          auto deficit = threshold - topic_stats->mesh_message_deliveries;
+          topic_stats->mesh_failure_penalty += deficit * deficit;
+        }
+        topic_stats->mesh_message_deliveries_active = false;
+        topic_stats->mesh_active.reset();
+      }
+    }
+    void duplicateMessage(const PeerId &peer_id,
+                          const MessageId &msg_id,
+                          const TopicId &topic) {
+      auto &record = deliveries_.getOrDefault(msg_id);
+      if (record.peers.contains(peer_id)) {
+        return;
+      }
+      if (std::holds_alternative<DeliveryStatusUnknown>(record.status)) {
+        record.peers.emplace(peer_id);
+      } else if (auto *valid =
+                     std::get_if<DeliveryStatusValid>(&record.status)) {
+        record.peers.emplace(peer_id);
+        markDuplicateMessageDelivery(peer_id, topic, valid->time);
+      } else if (std::holds_alternative<DeliveryStatusInvalid>(record.status)) {
+        markInvalidMessageDelivery(peer_id, topic);
+      }
+    }
+    void validateMessage(const PeerId &peer_id,
+                         const MessageId &msg_id,
+                         const TopicId &topic) {
+      deliveries_.getOrDefault(msg_id);
+    }
+    void connect(const PeerId &peer_id) {
+      peer_stats_[peer_id].expires_at.reset();
+    }
+    void disconnect(const PeerId &peer_id) {
+      auto it = peer_stats_.find(peer_id);
+      if (it == peer_stats_.end()) {
+        return;
+      }
+      if (score(peer_id) > 0) {
+        peer_stats_.erase(peer_id);
+        return;
+      }
+      for (auto &[topic, topic_stats] : it->second.topics) {
+        topic_stats.first_message_deliveries = 0;
+        auto topic_it = params_.topics.find(topic);
+        if (topic_it != params_.topics.end()) {
+          auto threshold = topic_it->second.mesh_message_deliveries_threshold;
+          if (topic_stats.mesh_active.has_value()
+              and topic_stats.mesh_message_deliveries_active
+              and topic_stats.mesh_message_deliveries < threshold) {
+            auto deficit = threshold - topic_stats.mesh_message_deliveries;
+            topic_stats.mesh_failure_penalty += deficit * deficit;
+          }
+        }
+        topic_stats.mesh_active.reset();
+        topic_stats.mesh_message_deliveries_active = false;
+      }
+      it->second.expires_at = Clock::now() + params_.retain_score;
+    }
+
+    void onDecay() {
+      auto now = Clock::now();
+      for (auto it = peer_stats_.begin(); it != peer_stats_.end();) {
+        auto &peer_stats = it->second;
+        if (it->second.expires_at.has_value()) {
+          if (*it->second.expires_at < now) {
+            it = peer_stats_.erase(it);
+          } else {
+            ++it;
+          }
+        } else {
+          ++it;
+          for (auto &[topic, topic_stats] : peer_stats.topics) {
+            auto topic_it = params_.topics.find(topic);
+            if (topic_it == params_.topics.end()) {
+              continue;
+            }
+            auto &topic_params = topic_it->second;
+            topic_stats.first_message_deliveries *=
+                topic_params.first_message_deliveries_decay;
+            if (topic_stats.first_message_deliveries < params_.decay_to_zero) {
+              topic_stats.first_message_deliveries = 0.0;
+            }
+            topic_stats.mesh_message_deliveries *=
+                topic_params.mesh_message_deliveries_decay;
+            if (topic_stats.mesh_message_deliveries < params_.decay_to_zero) {
+              topic_stats.mesh_message_deliveries = 0.0;
+            }
+            topic_stats.mesh_failure_penalty *=
+                topic_params.mesh_failure_penalty_decay;
+            if (topic_stats.mesh_failure_penalty < params_.decay_to_zero) {
+              topic_stats.mesh_failure_penalty = 0.0;
+            }
+            topic_stats.invalid_message_deliveries *=
+                topic_params.invalid_message_deliveries_decay;
+            if (topic_stats.invalid_message_deliveries
+                < params_.decay_to_zero) {
+              topic_stats.invalid_message_deliveries = 0.0;
+            }
+            if (topic_stats.mesh_active.has_value()) {
+              topic_stats.mesh_active->mesh_time =
+                  std::chrono::duration_cast<Duration>(
+                      now - topic_stats.mesh_active->graft_time);
+              if (topic_stats.mesh_active->mesh_time
+                  > topic_params.mesh_message_deliveries_activation) {
+                topic_stats.mesh_message_deliveries_active = true;
+              }
+            }
+          }
+          peer_stats.behaviour_penalty *= params_.behaviour_penalty_decay;
+          if (peer_stats.behaviour_penalty < params_.decay_to_zero) {
+            peer_stats.behaviour_penalty = 0;
+          }
+          peer_stats.slow_peer_penalty *= params_.slow_peer_decay;
+          if (peer_stats.slow_peer_penalty < params_.decay_to_zero) {
+            peer_stats.slow_peer_penalty = 0;
+          }
+        }
+      }
+    }
+
+   private:
+    TopicStats *statsOrDefault(PeerStats &peer, const TopicId &topic) {
+      auto it = peer.topics.find(topic);
+      if (it == peer.topics.end() and params_.topics.contains(topic)) {
+        it = peer.topics.emplace(topic, TopicStats{}).first;
+      }
+      return it == peer.topics.end() ? nullptr : &it->second;
+    }
+
+    void markDuplicateMessageDelivery(const PeerId &peer_id,
+                                      const TopicId &topic,
+                                      std::optional<Time> validated_time) {
+      auto it = peer_stats_.find(peer_id);
+      if (it == peer_stats_.end()) {
+        return;
+      }
+      if (auto *topic_stats = statsOrDefault(it->second, topic)) {
+        if (topic_stats->mesh_active) {
+          auto &topic_params = params_.topics.at(topic);
+          if (validated_time) {
+            auto now = Clock::now();
+            auto window_time =
+                *validated_time + topic_params.mesh_message_deliveries_window;
+            if (now > window_time) {
+              return;
+            }
+            topic_stats->mesh_message_deliveries =
+                std::min(topic_stats->mesh_message_deliveries + 1,
+                         topic_params.mesh_message_deliveries_cap);
+          }
+        }
+      }
+    }
+
+    void markInvalidMessageDelivery(const PeerId &peer_id,
+                                    const TopicId &topic) {
+      auto it = peer_stats_.find(peer_id);
+      if (it == peer_stats_.end()) {
+        return;
+      }
+      if (auto *topic_stats = statsOrDefault(it->second, topic)) {
+        topic_stats->invalid_message_deliveries += 1;
+      }
+    }
+
+    ScoreConfig params_;
+    std::unordered_map<PeerId, PeerStats> peer_stats_;
+    TimeCache<MessageId, DeliveryRecord, qtils::BytesStdHash> deliveries_;
+  };
+}  // namespace libp2p::protocol::gossip::score
+
+namespace libp2p::protocol::gossip {
+  using score::Score;
+}  // namespace libp2p::protocol::gossip
diff --git a/include/libp2p/protocol/gossip/score_config.hpp b/include/libp2p/protocol/gossip/score_config.hpp
new file mode 100644
index 000000000..63a705731
--- /dev/null
+++ b/include/libp2p/protocol/gossip/score_config.hpp
@@ -0,0 +1,85 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <chrono>
+#include <string>
+#include <unordered_map>
+
+namespace libp2p::protocol::gossip {
+  using TopicId = std::string;
+
+  constexpr double kDefaultDecayToZero = 0.1;
+
+  struct TopicScoreParams {
+    double topic_weight = 0.5;
+    double time_in_mesh_weight = 1;
+    std::chrono::milliseconds time_in_mesh_quantum{1};
+    double time_in_mesh_cap = 3600;
+    double first_message_deliveries_weight = 1;
+    double first_message_deliveries_decay = 0.5;
+    double first_message_deliveries_cap = 2000;
+    double mesh_message_deliveries_weight = -1;
+    double mesh_message_deliveries_decay = 0.5;
+    double mesh_message_deliveries_cap = 100;
+    double mesh_message_deliveries_threshold = 20;
+    std::chrono::milliseconds mesh_message_deliveries_window{10};
+    std::chrono::seconds mesh_message_deliveries_activation{5};
+    double mesh_failure_penalty_weight = -1;
+    double mesh_failure_penalty_decay = 0.5;
+    double invalid_message_deliveries_weight = -1;
+    double invalid_message_deliveries_decay = 0.3;
+  };
+
+  struct ScoreConfig {
+    double zero = 0;
+    /// The score threshold below which gossip propagation is suppressed;
+    /// should be negative.
+    double gossip_threshold = -10;
+    /// The score threshold below which we shouldn't publish when using flood
+    /// publishing (also applies to fanout peers); should be negative and <=
+    /// `gossip_threshold`.
+    double publish_threshold = -50;
+    /// The score threshold below which message processing is suppressed
+    /// altogether, implementing an effective graylist according to peer score;
+    /// should be negative and
+    /// <= `publish_threshold`.
+    double graylist_threshold = -80;
+    /// The median mesh score threshold before triggering opportunistic
+    /// grafting; this should have a small positive value.
+    double opportunistic_graft_threshold = 20;
+
+    std::unordered_map<TopicId, TopicScoreParams> topics;
+    double topic_score_cap = 3600;
+    double app_specific_weight = 10;
+    double behaviour_penalty_weight = -10;
+    double behaviour_penalty_threshold = 0;
+    double behaviour_penalty_decay = 0.2;
+    std::chrono::milliseconds decay_interval = std::chrono::seconds{1};
+    double decay_to_zero = kDefaultDecayToZero;
+    std::chrono::seconds retain_score{3600};
+    double slow_peer_weight = -0.2;
+    double slow_peer_threshold = 0.0;
+    double slow_peer_decay = 0.2;
+
+    bool valid() const {
+      if (gossip_threshold > 0) {
+        return false;
+      }
+      if (publish_threshold > 0 or publish_threshold > gossip_threshold) {
+        return false;
+      }
+      if (graylist_threshold > 0 or graylist_threshold > publish_threshold) {
+        return false;
+      }
+      if (opportunistic_graft_threshold < 0) {
+        return false;
+      }
+      return true;
+    }
+  };
+}  // namespace libp2p::protocol::gossip
diff --git a/include/libp2p/protocol/gossip/time_cache.hpp b/include/libp2p/protocol/gossip/time_cache.hpp
new file mode 100644
index 000000000..0e97c9498
--- /dev/null
+++ b/include/libp2p/protocol/gossip/time_cache.hpp
@@ -0,0 +1,187 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <chrono>
+#include <deque>
+#include <libp2p/peer/peer_id.hpp>
+#include <qtils/bytes_std_hash.hpp>
+#include <qtils/empty.hpp>
+#include <unordered_map>
+#include <unordered_set>
+
+namespace libp2p::protocol::gossip {
+  using MessageId = Bytes;
+}  // namespace libp2p::protocol::gossip
+
+namespace libp2p::protocol::gossip::time_cache {
+  using Ttl = std::chrono::milliseconds;
+  using Clock = std::chrono::steady_clock;
+  using Time = Clock::time_point;
+
+  /// This implements a time-based LRU cache for checking gossipsub message
+  /// duplicates.
+  template <typename K, typename V, typename H = std::hash<K>>
+  class TimeCache {
+   public:
+    TimeCache(Ttl ttl) : ttl_{ttl} {}
+
+    size_t size() const {
+      return map_.size();
+    }
+
+    bool contains(const K &key) const {
+      return map_.contains(key);
+    }
+
+    void clearExpired(Time now = Clock::now()) {
+      while (not expirations_.empty() and expirations_.front().first <= now) {
+        map_.erase(expirations_.front().second);
+        expirations_.pop_front();
+      }
+    }
+
+    V &getOrDefault(const K &key, Time now = Clock::now()) {
+      clearExpired(now);
+      auto it = map_.find(key);
+      if (it == map_.end()) {
+        it = map_.emplace(key, V()).first;
+        expirations_.emplace_back(now + ttl_, it);
+      }
+      return it->second;
+    }
+
+    void popFront() {
+      if (expirations_.empty()) {
+        throw std::logic_error{"TimeCache::popFront empty"};
+      }
+      map_.erase(expirations_.front().second);
+      expirations_.pop_front();
+    }
+
+   private:
+    using Map = std::unordered_map<K, V>;
+
+    Ttl ttl_;
+    Map map_;
+    std::deque<std::pair<Clock::time_point, typename Map::iterator>>
+        expirations_;
+  };
+
+  template <typename K, typename H = std::hash<K>>
+  class DuplicateCache {
+   public:
+    DuplicateCache(Ttl ttl) : cache_{ttl} {}
+
+    bool contains(const K &key) const {
+      return cache_.contains(key);
+    }
+
+    bool insert(const K &key, Time now = Clock::now()) {
+      cache_.clearExpired(now);
+      if (cache_.contains(key)) {
+        return false;
+      }
+      cache_.getOrDefault(key);
+      return true;
+    }
+
+   private:
+    TimeCache<K, qtils::Empty, H> cache_;
+  };
+
+  template <typename K, typename H = std::hash<K>>
+  class IDontWantCache {
+    static constexpr size_t kCapacity = 10000;
+    static constexpr Ttl kTtl = std::chrono::seconds{3};
+
+   public:
+    void clearExpired(Time now = Clock::now()) {
+      cache_.clearExpired(now);
+    }
+
+    bool contains(const K &key) const {
+      return cache_.contains(key);
+    }
+
+    void insert(const K &key) {
+      if (cache_.contains(key)) {
+        return;
+      }
+      if (cache_.size() >= kCapacity) {
+        cache_.popFront();
+      }
+      cache_.getOrDefault(key);
+    }
+
+   private:
+    TimeCache<K, qtils::Empty, H> cache_{kTtl};
+  };
+
+  class GossipPromises {
+   public:
+    GossipPromises(Ttl ttl) : ttl_{ttl} {}
+
+    bool contains(const MessageId &message_id) const {
+      return map_.contains(message_id);
+    }
+
+    void add(const MessageId &message_id,
+             const PeerId &peer_id,
+             Time now = Clock::now()) {
+      map_[message_id].emplace(peer_id, now + ttl_);
+    }
+
+    void remove(const MessageId &message_id) {
+      map_.erase(message_id);
+    }
+
+    void peers(const MessageId &message_id, const auto &f) {
+      auto it = map_.find(message_id);
+      if (it != map_.end()) {
+        for (auto &p : it->second) {
+          f(p.first);
+        }
+      }
+    }
+
+    auto clearExpired(Time now = Clock::now()) {
+      std::unordered_map<PeerId, size_t> result;
+      for (auto it1 = map_.begin(); it1 != map_.end();) {
+        auto &map2 = it1->second;
+        for (auto it2 = map2.begin(); it2 != map2.end();) {
+          if (it2->second < now) {
+            ++result[it2->first];
+            it2 = map2.erase(it2);
+          } else {
+            ++it2;
+          }
+        }
+        if (map2.empty()) {
+          it1 = map_.erase(it1);
+        } else {
+          ++it1;
+        }
+      }
+      return result;
+    }
+
+   private:
+    Ttl ttl_;
+    std::unordered_map<MessageId,
+                       std::unordered_map<PeerId, Time>,
+                       qtils::BytesStdHash>
+        map_;
+  };
+}  // namespace libp2p::protocol::gossip::time_cache
+
+namespace libp2p::protocol::gossip {
+  using time_cache::DuplicateCache;
+  using time_cache::GossipPromises;
+  using time_cache::IDontWantCache;
+  using time_cache::TimeCache;
+}  // namespace libp2p::protocol::gossip
diff --git a/src/basic/message_read_writer_uvarint.cpp b/src/basic/message_read_writer_uvarint.cpp
index 26871ca36..6bc2b2b02 100644
--- a/src/basic/message_read_writer_uvarint.cpp
+++ b/src/basic/message_read_writer_uvarint.cpp
@@ -32,8 +32,8 @@ namespace libp2p::basic {
           }
 
           auto msg_len = varint_res.value().toUInt64();
+          auto buffer = std::make_shared<std::vector<uint8_t>>(msg_len, 0);
           if (0 != msg_len) {
-            auto buffer = std::make_shared<std::vector<uint8_t>>(msg_len, 0);
             self->conn_->read(
                 *buffer,
                 msg_len,
@@ -44,7 +44,7 @@ namespace libp2p::basic {
                   cb(std::move(buffer));
                 });
           } else {
-            cb(ResultType{});
+            cb(buffer);
           }
         });
   }
diff --git a/src/connection/CMakeLists.txt b/src/connection/CMakeLists.txt
index da0ca9d1a..59a0563ac 100644
--- a/src/connection/CMakeLists.txt
+++ b/src/connection/CMakeLists.txt
@@ -19,5 +19,12 @@ target_link_libraries(p2p_loopback_stream
     p2p_peer_id
     )
 
+libp2p_add_library(p2p_stream_pair
+    stream_pair.cpp
+    )
+target_link_libraries(p2p_stream_pair
+    p2p_peer_id
+    )
+
 libp2p_install(p2p_loopback_stream)
 
diff --git a/src/connection/stream_pair.cpp b/src/connection/stream_pair.cpp
new file mode 100644
index 000000000..a67184deb
--- /dev/null
+++ b/src/connection/stream_pair.cpp
@@ -0,0 +1,174 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <libp2p/connection/stream_pair.hpp>
+
+#include <boost/asio/streambuf.hpp>
+#include <libp2p/basic/read_return_size.hpp>
+#include <libp2p/basic/scheduler.hpp>
+#include <libp2p/common/ambigous_size.hpp>
+#include <libp2p/connection/stream.hpp>
+#include <qtils/option_take.hpp>
+
+namespace libp2p::connection {
+  class StreamPair : public std::enable_shared_from_this<StreamPair>,
+                     public Stream {
+    friend std::pair<std::shared_ptr<Stream>, std::shared_ptr<Stream>>
+    streamPair(std::shared_ptr<basic::Scheduler> post,
+               PeerId peer1,
+               PeerId peer2);
+
+   public:
+    StreamPair(std::shared_ptr<basic::Scheduler> post,
+               bool is_initiator,
+               PeerId peer_id)
+        : post_{std::move(post)},
+          is_initiator_{is_initiator},
+          peer_id_{std::move(peer_id)} {}
+
+    // clang-tidy cppcoreguidelines-special-member-functions
+    StreamPair(const StreamPair &) = delete;
+    void operator=(const StreamPair &) = delete;
+    StreamPair(StreamPair &&) = delete;
+    void operator=(StreamPair &&) = delete;
+
+    ~StreamPair() override {
+      reset();
+    }
+
+    void read(BytesOut out, size_t bytes, ReadCallbackFunc cb) override {
+      ambigousSize(out, bytes);
+      libp2p::readReturnSize(shared_from_this(), out, std::move(cb));
+    }
+    void readSome(BytesOut out, size_t bytes, ReadCallbackFunc cb) override {
+      ambigousSize(out, bytes);
+      if (out.empty()) {
+        throw std::logic_error{"StreamPair::readSome zero bytes"};
+      }
+      if (read_buffer_.size() == 0) {
+        if (read_closed_) {
+          post(std::move(cb), make_error_code(boost::asio::error::eof));
+          return;
+        }
+        reading_ = Reading{out, std::move(cb)};
+        return;
+      }
+      auto n = std::min(read_buffer_.size(), out.size());
+      memcpy(out.data(), read_buffer_.data().data(), n);
+      read_buffer_.consume(n);
+      post(std::move(cb), n);
+    }
+    void deferReadCallback(outcome::result<size_t> res,
+                           ReadCallbackFunc cb) override {
+      post(std::move(cb), res);
+    }
+
+    void writeSome(BytesIn in, size_t bytes, WriteCallbackFunc cb) override {
+      ambigousSize(in, bytes);
+      if (in.empty()) {
+        throw std::logic_error{"StreamPair::writeSome zero bytes"};
+      }
+      if (auto writer = writer_.lock()) {
+        writer->onRead(in);
+        post(std::move(cb), in.size());
+      } else {
+        post(std::move(cb), std::errc::broken_pipe);
+      }
+    }
+    void deferWriteCallback(std::error_code ec, WriteCallbackFunc cb) override {
+      post(std::move(cb), ec);
+    }
+
+    bool isClosedForRead() const override {
+      return read_buffer_.size() == 0 and read_closed_;
+    }
+    bool isClosedForWrite() const override {
+      return writer_.expired();
+    }
+    bool isClosed() const override {
+      return isClosedForRead() and isClosedForWrite();
+    }
+    void close(VoidResultHandlerFunc cb) override {
+      if (auto writer = writer_.lock()) {
+        writer->onReadClose();
+      }
+      onWriteClose();
+      post(std::move(cb), outcome::success());
+    }
+    void reset() override {
+      if (auto writer = writer_.lock()) {
+        writer->onReadClose();
+        writer->onWriteClose();
+      }
+      onReadClose();
+      onWriteClose();
+    }
+    void adjustWindowSize(uint32_t, VoidResultHandlerFunc) override {}
+    outcome::result<bool> isInitiator() const override {
+      return is_initiator_;
+    }
+    outcome::result<PeerId> remotePeerId() const override {
+      return peer_id_;
+    }
+    outcome::result<Multiaddress> localMultiaddr() const override {
+      throw std::logic_error{"StreamPair::localMultiaddr"};
+    }
+    outcome::result<Multiaddress> remoteMultiaddr() const override {
+      return Multiaddress::create(fmt::format("/p2p/{}", peer_id_.toBase58()))
+          .value();
+    }
+
+   private:
+    void post(auto cb, auto arg) {
+      post_->schedule([cb{std::move(cb)}, arg{std::move(arg)}]() mutable {
+        cb(std::move(arg));
+      });
+    }
+    void onRead(BytesIn in) {
+      if (auto reading = qtils::optionTake(reading_)) {
+        auto n = std::min(in.size(), reading->out.size());
+        memcpy(reading->out.data(), in.data(), n);
+        in = in.subspan(n);
+        post(std::move(reading->cb), n);
+      }
+      if (not in.empty()) {
+        auto n = in.size();
+        memcpy(read_buffer_.prepare(n).data(), in.data(), n);
+        read_buffer_.commit(n);
+      }
+    }
+    void onReadClose() {
+      read_closed_ = true;
+      if (auto reading = qtils::optionTake(reading_)) {
+        post(std::move(reading->cb), make_error_code(boost::asio::error::eof));
+      }
+    }
+    void onWriteClose() {
+      writer_.reset();
+    }
+
+    struct Reading {
+      BytesOut out;
+      ReadCallbackFunc cb;
+    };
+    std::shared_ptr<basic::Scheduler> post_;
+    bool is_initiator_;
+    PeerId peer_id_;
+    std::weak_ptr<StreamPair> writer_;
+    std::optional<Reading> reading_;
+    boost::asio::streambuf read_buffer_;
+    bool read_closed_ = false;
+  };
+
+  std::pair<std::shared_ptr<Stream>, std::shared_ptr<Stream>> streamPair(
+      std::shared_ptr<basic::Scheduler> post, PeerId peer1, PeerId peer2) {
+    auto stream1 = std::make_shared<StreamPair>(post, true, peer2);
+    auto stream2 = std::make_shared<StreamPair>(post, false, peer1);
+    stream1->writer_ = stream2;
+    stream2->writer_ = stream1;
+    return {stream1, stream2};
+  }
+}  // namespace libp2p::connection
diff --git a/src/protocol/gossip/impl/choose_peers.hpp b/src/protocol/gossip/impl/choose_peers.hpp
new file mode 100644
index 000000000..906fb0205
--- /dev/null
+++ b/src/protocol/gossip/impl/choose_peers.hpp
@@ -0,0 +1,50 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <deque>
+#include <random>
+
+#include "peer_context.hpp"
+
+namespace libp2p::protocol::gossip {
+  class ChoosePeers {
+   public:
+    std::deque<PeerContextPtr> choose(
+        auto &&all,
+        const std::invocable<PeerContextPtr> auto &predicate,
+        const std::invocable<size_t> auto &get_count) {
+      std::deque<PeerContextPtr> chosen;
+      for (auto &ctx : all) {
+        if (ctx->isGossipsub() and predicate(ctx)) {
+          chosen.emplace_back(ctx);
+        }
+      }
+      std::ranges::shuffle(chosen, random_);
+      auto count = get_count(chosen.size());
+      if (chosen.size() > count) {
+        chosen.resize(count);
+      }
+      return chosen;
+    }
+    std::deque<PeerContextPtr> choose(
+        auto &&all,
+        const std::invocable<PeerContextPtr> auto &predicate,
+        size_t count) {
+      return choose(std::forward<decltype(all)>(all), predicate, [&](size_t) {
+        return count;
+      });
+    }
+
+    void shuffle(auto &&r) {
+      std::ranges::shuffle(std::forward<decltype(r)>(r), random_);
+    }
+
+   private:
+    std::default_random_engine random_;
+  };
+}  // namespace libp2p::protocol::gossip
diff --git a/src/protocol/gossip/impl/common.hpp b/src/protocol/gossip/impl/common.hpp
index 0097aebfa..79bffa2d2 100644
--- a/src/protocol/gossip/impl/common.hpp
+++ b/src/protocol/gossip/impl/common.hpp
@@ -8,6 +8,7 @@
 
 #include <cstdint>
 #include <memory>
+#include <unordered_set>
 
 #include <libp2p/protocol/gossip/gossip.hpp>
 
@@ -47,6 +48,8 @@ namespace libp2p::protocol::gossip {
   /// Remote peer and its context
   using PeerContextPtr = std::shared_ptr<struct PeerContext>;
 
+  using OutboundPeers = std::unordered_set<PeerId>;
+
   /// Message being published
   struct TopicMessage {
     using Ptr = std::shared_ptr<TopicMessage>;
diff --git a/src/protocol/gossip/impl/connectivity.cpp b/src/protocol/gossip/impl/connectivity.cpp
index 8a308d64d..08d165044 100644
--- a/src/protocol/gossip/impl/connectivity.cpp
+++ b/src/protocol/gossip/impl/connectivity.cpp
@@ -29,7 +29,7 @@ namespace libp2p::protocol::gossip {
   Connectivity::Connectivity(Config config,
                              std::shared_ptr<basic::Scheduler> scheduler,
                              std::shared_ptr<Host> host,
-                             std::shared_ptr<MessageReceiver> msg_receiver,
+                             std::weak_ptr<MessageReceiver> msg_receiver,
                              ConnectionStatusFeedback on_connected)
       : config_(std::move(config)),
         scheduler_(std::move(scheduler)),
@@ -38,7 +38,16 @@ namespace libp2p::protocol::gossip {
         connected_cb_(std::move(on_connected)),
         log_("gossip",
              "Connectivity",
-             host_->getPeerInfo().id.toBase58().substr(46)) {}
+             host_->getPeerInfo().id.toBase58().substr(46)) {
+    for (auto &p : config_.protocol_versions) {
+      protocols_.emplace_back(p.first);
+    }
+    std::ranges::sort(protocols_,
+                      [&](const ProtocolName &l, const ProtocolName &r) {
+                        return config_.protocol_versions.at(l)
+                             > config_.protocol_versions.at(r);
+                      });
+  }
 
   Connectivity::~Connectivity() {
     stop();
@@ -58,7 +67,7 @@ namespace libp2p::protocol::gossip {
         };
 
     host_->setProtocolHandler(
-        {config_.protocol_version},
+        protocols_,
         [self_wptr=weak_from_this()]
             (StreamAndProtocol stream) {
           auto h = self_wptr.lock();
@@ -134,10 +143,6 @@ namespace libp2p::protocol::gossip {
     ctx->outbound_stream->write(std::move(serialized));
   }
 
-  peer::ProtocolName Connectivity::getProtocolId() const {
-    return config_.protocol_version;
-  }
-
   void Connectivity::handle(StreamAndProtocol stream_and_protocol) {
     auto &stream = stream_and_protocol.stream;
 
@@ -178,6 +183,7 @@ namespace libp2p::protocol::gossip {
         unban(ctx);
       }
     }
+    updatePeerKind(*ctx, stream_and_protocol);
 
     size_t stream_id = 0;
     bool is_new_connection = false;
@@ -189,7 +195,7 @@ namespace libp2p::protocol::gossip {
                                                   config_,
                                                   *scheduler_,
                                                   on_stream_event_,
-                                                  *msg_receiver_,
+                                                  msg_receiver_,
                                                   std::move(stream),
                                                   ctx);
 
@@ -232,7 +238,7 @@ namespace libp2p::protocol::gossip {
     // clang-format off
     host_->newStream(
         pi,
-        {config_.protocol_version},
+        protocols_,
         [wptr = weak_from_this(), this, ctx=ctx] (auto &&rstream) mutable {
             auto self = wptr.lock();
           if (self) {
@@ -253,7 +259,7 @@ namespace libp2p::protocol::gossip {
     // clang-format off
     host_->newStream(
         ctx->peer_id,
-        {config_.protocol_version},
+        protocols_,
         [wptr = weak_from_this(), this, ctx=ctx] (auto &&rstream) mutable {
           auto self = wptr.lock();
           if (self) {
@@ -279,6 +285,7 @@ namespace libp2p::protocol::gossip {
     }
 
     auto &stream = rstream.value().stream;
+    updatePeerKind(*ctx, rstream.value());
 
     if (!started_) {
       stream->reset();
@@ -306,7 +313,7 @@ namespace libp2p::protocol::gossip {
                                                   config_,
                                                   *scheduler_,
                                                   on_stream_event_,
-                                                  *msg_receiver_,
+                                                  msg_receiver_,
                                                   std::move(stream),
                                                   ctx);
 
@@ -395,26 +402,31 @@ namespace libp2p::protocol::gossip {
     banOrForget(from);
   }
 
-  void Connectivity::peerIsWritable(const PeerContextPtr &ctx,
-                                    bool low_latency) {
+  void Connectivity::peerIsWritable(const PeerContextPtr &ctx) {
     if (ctx->message_builder->empty()) {
       return;
     }
 
-    if (low_latency) {
-      writable_peers_low_latency_.insert(ctx);
-    } else {
-      writable_peers_on_heartbeat_.insert(ctx);
+    const auto was_empty = writable_peers_.empty();
+    writable_peers_.insert(ctx);
+    if (was_empty) {
+      scheduler_->schedule([weak_self{weak_from_this()}] {
+        auto self = weak_self.lock();
+        if (not self) {
+          return;
+        }
+        self->flush();
+      });
     }
   }
 
   void Connectivity::flush() {
-    writable_peers_low_latency_.selectAll(
+    writable_peers_.selectAll(
         [this](const PeerContextPtr &ctx) { flush(ctx); });
-    writable_peers_low_latency_.clear();
+    writable_peers_.clear();
   }
 
-  void Connectivity::onHeartbeat(const std::map<TopicId, bool> &local_changes) {
+  void Connectivity::onHeartbeat() {
     if (!started_) {
       return;
     }
@@ -448,32 +460,6 @@ namespace libp2p::protocol::gossip {
       }
     }
 
-    if (!local_changes.empty()) {
-      // we have something to say to all connected peers
-
-      std::vector<std::pair<bool, TopicId>> flat_changes;
-      flat_changes.reserve(local_changes.size());
-      boost::for_each(local_changes, [&flat_changes](auto &&p) {
-        flat_changes.emplace_back(p.second, std::move(p.first));
-      });
-
-      connected_peers_.selectAll(
-          [&flat_changes, this](const PeerContextPtr &ctx) {
-            boost::for_each(flat_changes, [&ctx](auto &&p) {
-              ctx->message_builder->addSubscription(p.first, p.second);
-            });
-            flush(ctx);
-          });
-
-    } else {
-      flush();
-      writable_peers_on_heartbeat_.selectAll(
-          [this](const PeerContextPtr &ctx) { flush(ctx); });
-    }
-
-    writable_peers_low_latency_.clear();
-    writable_peers_on_heartbeat_.clear();
-
     if (ts >= addresses_renewal_time_) {
       addresses_renewal_time_ =
           scheduler_->now() + config_.address_expiration_msec * 9 / 10;
@@ -489,4 +475,17 @@ namespace libp2p::protocol::gossip {
     return connected_peers_;
   }
 
+  void Connectivity::subscribe(const TopicId &topic, bool subscribe) {
+    for (auto &ctx : connected_peers_) {
+      ctx->message_builder->addSubscription(subscribe, topic);
+      peerIsWritable(ctx);
+    }
+  }
+
+  void Connectivity::updatePeerKind(PeerContext &ctx,
+                                    const StreamAndProtocol &stream) const {
+    if (not ctx.peer_kind) {
+      ctx.peer_kind = config_.protocol_versions.at(stream.protocol);
+    }
+  }
 }  // namespace libp2p::protocol::gossip
diff --git a/src/protocol/gossip/impl/connectivity.hpp b/src/protocol/gossip/impl/connectivity.hpp
index fe7a603a2..932ccdba0 100644
--- a/src/protocol/gossip/impl/connectivity.hpp
+++ b/src/protocol/gossip/impl/connectivity.hpp
@@ -21,8 +21,7 @@ namespace libp2p::protocol::gossip {
   class MessageReceiver;
 
   /// Part of GossipCore: Protocol server and network connections manager
-  class Connectivity : public protocol::BaseProtocol,
-                       public std::enable_shared_from_this<Connectivity> {
+  class Connectivity : public std::enable_shared_from_this<Connectivity> {
    public:
     Connectivity(const Connectivity &) = delete;
     Connectivity &operator=(const Connectivity &) = delete;
@@ -35,10 +34,10 @@ namespace libp2p::protocol::gossip {
     Connectivity(Config config,
                  std::shared_ptr<basic::Scheduler> scheduler,
                  std::shared_ptr<Host> host,
-                 std::shared_ptr<MessageReceiver> msg_receiver,
+                 std::weak_ptr<MessageReceiver> msg_receiver,
                  ConnectionStatusFeedback on_connected);
 
-    ~Connectivity() override;
+    ~Connectivity();
 
     void start();
 
@@ -52,26 +51,25 @@ namespace libp2p::protocol::gossip {
     /// Add peer to writable set, actual writes occur on flush() (piggybacking)
     /// The idea behind writable set and flush() is a compromise between
     /// latency and message rate
-    void peerIsWritable(const PeerContextPtr &ctx, bool low_latency);
+    void peerIsWritable(const PeerContextPtr &ctx);
 
     /// Flushes all pending writes for peers in writable set
     void flush();
 
     /// Performs periodic tasks and broadcasts heartbeat message to
     /// all connected peers. The changes are subscribe/unsubscribe events
-    void onHeartbeat(const std::map<TopicId, bool> &local_changes);
+    void onHeartbeat();
 
     /// Returns connected peers
     const PeerSet &getConnectedPeers() const;
 
+    void subscribe(const TopicId &topic, bool subscribe);
+
    private:
     using BannedPeers = std::set<std::pair<Time, PeerContextPtr>>;
 
-    /// BaseProtocol override
-    peer::ProtocolName getProtocolId() const override;
-
-    /// BaseProtocol override, on new inbound stream
-    void handle(StreamAndProtocol stream) override;
+    /// On new inbound stream
+    void handle(StreamAndProtocol stream);
 
     /// Tries to connect to peer
     void dial(const PeerContextPtr &peer);
@@ -99,10 +97,14 @@ namespace libp2p::protocol::gossip {
     /// Flushes outgoing messages into wire for a given peer, if connected
     void flush(const PeerContextPtr &ctx) const;
 
+    void updatePeerKind(PeerContext &ctx,
+                        const StreamAndProtocol &stream) const;
+
     const Config config_;
+    StreamProtocols protocols_;
     std::shared_ptr<basic::Scheduler> scheduler_;
     std::shared_ptr<Host> host_;
-    std::shared_ptr<MessageReceiver> msg_receiver_;
+    std::weak_ptr<MessageReceiver> msg_receiver_;
     ConnectionStatusFeedback connected_cb_;
     Stream::Feedback on_stream_event_;
     bool started_ = false;
@@ -120,11 +122,8 @@ namespace libp2p::protocol::gossip {
     /// Writable peers
     PeerSet connected_peers_;
 
-    /// Peers with pending write operation before the next heartbeat
-    PeerSet writable_peers_low_latency_;
-
-    /// Peers to be flushed on next heartbeat
-    PeerSet writable_peers_on_heartbeat_;
+    /// Peers with pending write operation
+    PeerSet writable_peers_;
 
     /// Renew addresses in address repo periodically within heartbeat timer
     std::chrono::milliseconds addresses_renewal_time_{0};
diff --git a/src/protocol/gossip/impl/gossip_core.cpp b/src/protocol/gossip/impl/gossip_core.cpp
index 2ee17239e..1b75e4990 100644
--- a/src/protocol/gossip/impl/gossip_core.cpp
+++ b/src/protocol/gossip/impl/gossip_core.cpp
@@ -11,6 +11,7 @@
 #include <libp2p/crypto/crypto_provider.hpp>
 #include <libp2p/crypto/key_marshaller.hpp>
 #include <libp2p/peer/identity_manager.hpp>
+#include <libp2p/protocol/gossip/score.hpp>
 #include <qtils/hex.hpp>
 
 #include "connectivity.hpp"
@@ -54,9 +55,13 @@ namespace libp2p::protocol::gossip {
         key_marshaller_(std::move(key_marshaller)),
         local_peer_id_(host_->getPeerInfo().id),
         msg_cache_(
-            config_.message_cache_lifetime_msec,
+            config_.history_length * config_.heartbeat_interval_msec,
             [sch = scheduler_] { return sch->now(); }
         ),
+        score_{std::make_shared<Score>(config_.score)},
+        outbound_peers_{std::make_shared<OutboundPeers>()},
+        duplicate_cache_{config.duplicate_cache_time},
+        gossip_promises_{config.iwant_followup_time},
         local_subscriptions_(std::make_shared<LocalSubscriptions>(
             [this](bool subscribe, const TopicId &topic) {
               onLocalSubscriptionChanged(subscribe, topic);
@@ -109,7 +114,7 @@ namespace libp2p::protocol::gossip {
     }
 
     remote_subscriptions_ = std::make_shared<RemoteSubscriptions>(
-        config_, *connectivity_, *scheduler_, log_);
+        config_, *connectivity_, score_, outbound_peers_, scheduler_, log_);
 
     started_ = true;
 
@@ -118,8 +123,40 @@ namespace libp2p::protocol::gossip {
     }
 
     setTimerHeartbeat();
+    setTimerScore();
 
     connectivity_->start();
+    new_connection_sub_ =
+        host_->getBus()
+            .getChannel<event::network::OnNewConnectionChannel>()
+            .subscribe(
+                [weak_self{weak_from_this()}](
+                    std::weak_ptr<connection::CapableConnection> weak_conn) {
+                  if (auto self = weak_self.lock()) {
+                    if (auto conn = weak_conn.lock()) {
+                      auto peer_id = conn->remotePeer().value();
+                      if (self->host_->getNetwork()
+                              .getConnectionManager()
+                              .getConnectionsToPeer(peer_id)
+                              .size()
+                          == 1) {
+                        self->score_->connect(peer_id);
+                        if (conn->isInitiator()) {
+                          self->outbound_peers_->insert(peer_id);
+                        }
+                      }
+                    }
+                  }
+                });
+    peer_disconnected_sub_ =
+        host_->getBus()
+            .getChannel<event::network::OnPeerDisconnectedChannel>()
+            .subscribe([weak_self{weak_from_this()}](const PeerId &peer_id) {
+              if (auto self = weak_self.lock()) {
+                self->score_->disconnect(peer_id);
+                self->outbound_peers_->erase(peer_id);
+              }
+            });
   }
 
   void GossipCore::stop() {
@@ -177,6 +214,10 @@ namespace libp2p::protocol::gossip {
 
     MessageId msg_id = create_message_id_(msg->from, msg->seq_no, msg->data);
 
+    if (not duplicate_cache_.insert(msg_id)) {
+      return false;
+    }
+
     [[maybe_unused]] bool inserted = msg_cache_.insert(msg, msg_id);
     assert(inserted);
 
@@ -223,25 +264,44 @@ namespace libp2p::protocol::gossip {
                            const MessageId &msg_id) {
     assert(started_);
 
+    if (not from->isGossipsub()) {
+      return;
+    }
+    if (score_->below(from->peer_id, config_.score.gossip_threshold)) {
+      return;
+    }
+
     log_.debug("peer {} has msg for topic {}", from->str, topic);
 
-    if (remote_subscriptions_->hasTopic(topic)
-        && !msg_cache_.contains(msg_id)) {
+    if (remote_subscriptions_->isSubscribed(topic)
+        and not duplicate_cache_.contains(msg_id)
+        and not gossip_promises_.contains(msg_id)) {
       log_.debug("requesting msg id {:x}", msg_id);
 
+      gossip_promises_.add(msg_id, from->peer_id);
       from->message_builder->addIWant(msg_id);
-      connectivity_->peerIsWritable(from, false);
+      connectivity_->peerIsWritable(from);
     }
   }
 
   void GossipCore::onIWant(const PeerContextPtr &from,
                            const MessageId &msg_id) {
+    if (not from->isGossipsub()) {
+      return;
+    }
+    if (score_->below(from->peer_id, config_.score.gossip_threshold)) {
+      return;
+    }
+    if (from->idontwant.contains(msg_id)) {
+      return;
+    }
+
     log_.debug("peer {} wants message {:x}", from->str, msg_id);
 
     auto msg_found = msg_cache_.getMessage(msg_id);
     if (msg_found) {
       from->message_builder->addMessage(*msg_found.value(), msg_id);
-      connectivity_->peerIsWritable(from, true);
+      connectivity_->peerIsWritable(from);
     } else {
       log_.debug("wanted message not in cache");
     }
@@ -250,6 +310,13 @@ namespace libp2p::protocol::gossip {
   void GossipCore::onGraft(const PeerContextPtr &from, const TopicId &topic) {
     assert(started_);
 
+    if (not from->isGossipsub()) {
+      return;
+    }
+    if (score_->below(from->peer_id, config_.score.graylist_threshold)) {
+      return;
+    }
+
     log_.debug("graft from peer {} for topic {}", from->str, topic);
 
     remote_subscriptions_->onGraft(from, topic);
@@ -257,9 +324,16 @@ namespace libp2p::protocol::gossip {
 
   void GossipCore::onPrune(const PeerContextPtr &from,
                            const TopicId &topic,
-                           uint64_t backoff_time) {
+                           std::optional<std::chrono::seconds> backoff_time) {
     assert(started_);
 
+    if (not from->isGossipsub()) {
+      return;
+    }
+    if (score_->below(from->peer_id, config_.score.graylist_threshold)) {
+      return;
+    }
+
     log_.debug("prune from peer {} for topic {}", from->str, topic);
 
     remote_subscriptions_->onPrune(from, topic, backoff_time);
@@ -269,22 +343,39 @@ namespace libp2p::protocol::gossip {
                                   TopicMessage::Ptr msg) {
     assert(started_);
 
+    if (score_->below(from->peer_id, config_.score.graylist_threshold)) {
+      return;
+    }
+
     // do we need this message?
-    auto subscribed = remote_subscriptions_->hasTopic(msg->topic);
+    auto subscribed = remote_subscriptions_->isSubscribed(msg->topic);
     if (!subscribed) {
       // ignore this message
       return;
     }
 
     MessageId msg_id = create_message_id_(msg->from, msg->seq_no, msg->data);
-    log_.debug("message arrived, msg id={:x}", msg_id);
 
-    if (msg_cache_.contains(msg_id)) {
-      // already there, ignore
-      log_.debug("ignoring message, already in cache");
+    if (MessageBuilder::pbSize(*msg)
+        > config_.idontwant_message_size_threshold) {
+      gossip_promises_.peers(msg_id, [&](const PeerId &peer_id) {
+        if (auto ctx_opt = connectivity_->getConnectedPeers().find(peer_id)) {
+          auto &ctx = *ctx_opt;
+          if (ctx->isGossipsubv1_2()) {
+            ctx->message_builder->addIDontWant(msg_id);
+            connectivity_->peerIsWritable(ctx);
+          }
+        }
+      });
+    }
+
+    if (not duplicate_cache_.insert(msg_id)) {
+      score_->duplicateMessage(from->peer_id, msg_id, msg->topic);
       return;
     }
 
+    log_.debug("message arrived, msg id={:x}", msg_id);
+
     // validate message. If no validator is set then we
     // suppose that the message is valid (we might not know topic details)
     bool valid = true;
@@ -301,6 +392,10 @@ namespace libp2p::protocol::gossip {
       return;
     }
 
+    score_->validateMessage(from->peer_id, msg_id, msg->topic);
+
+    gossip_promises_.remove(msg_id);
+
     if (!msg_cache_.insert(msg, msg_id)) {
       log_.error("message cache error");
       return;
@@ -312,18 +407,32 @@ namespace libp2p::protocol::gossip {
     remote_subscriptions_->onNewMessage(from, msg, msg_id);
   }
 
+  void GossipCore::onIDontWant(const PeerContextPtr &from,
+                               const std::vector<MessageId> &message_ids) {
+    if (not from->isGossipsubv1_2()) {
+      return;
+    }
+    if (score_->below(from->peer_id, config_.score.graylist_threshold)) {
+      return;
+    }
+    for (auto &message_id : message_ids) {
+      from->idontwant.insert(message_id);
+    }
+  }
+
   void GossipCore::onMessageEnd(const PeerContextPtr &from) {
     assert(started_);
 
     log_.debug("finished dispatching message from peer {}", from->str);
-
-    // Apply immediate send operation to affected peers
-    connectivity_->flush();
   }
 
   void GossipCore::onHeartbeat() {
     assert(started_);
 
+    for (auto &[peer_id, count] : gossip_promises_.clearExpired()) {
+      score_->addPenalty(peer_id, count);
+    }
+
     // shift cache
     msg_cache_.shift();
 
@@ -331,8 +440,7 @@ namespace libp2p::protocol::gossip {
     remote_subscriptions_->onHeartbeat();
 
     // send changes to peers
-    connectivity_->onHeartbeat(broadcast_on_heartbeat_);
-    broadcast_on_heartbeat_.clear();
+    connectivity_->onHeartbeat();
 
     setTimerHeartbeat();
   }
@@ -347,8 +455,7 @@ namespace libp2p::protocol::gossip {
         for (const auto &local_sub : local_subscriptions_->subscribedTo()) {
           ctx->message_builder->addSubscription(true, local_sub.first);
         }
-        connectivity_->peerIsWritable(ctx, true);
-        connectivity_->flush();
+        connectivity_->peerIsWritable(ctx);
       }
     } else {
       log_.debug("peer {} disconnected", ctx->str);
@@ -362,14 +469,7 @@ namespace libp2p::protocol::gossip {
       return;
     }
 
-    // send this notification on next heartbeat to all connected peers
-    auto it = broadcast_on_heartbeat_.find(topic);
-    if (it == broadcast_on_heartbeat_.end()) {
-      broadcast_on_heartbeat_.emplace(topic, subscribe);
-    } else if (it->second != subscribe) {
-      // save traffic
-      broadcast_on_heartbeat_.erase(it);
-    }
+    connectivity_->subscribe(topic, subscribe);
 
     // update meshes per topic
     remote_subscriptions_->onSelfSubscribed(subscribe, topic);
@@ -386,4 +486,17 @@ namespace libp2p::protocol::gossip {
         },
         config_.heartbeat_interval_msec);
   }
+
+  void GossipCore::setTimerScore() {
+    scheduler_->schedule(
+        [weak_self{weak_from_this()}] {
+          auto self = weak_self.lock();
+          if (not self) {
+            return;
+          }
+          self->score_->onDecay();
+          self->setTimerScore();
+        },
+        config_.score.decay_interval);
+  }
 }  // namespace libp2p::protocol::gossip
diff --git a/src/protocol/gossip/impl/gossip_core.hpp b/src/protocol/gossip/impl/gossip_core.hpp
index 0d7e42509..3239a7bf9 100644
--- a/src/protocol/gossip/impl/gossip_core.hpp
+++ b/src/protocol/gossip/impl/gossip_core.hpp
@@ -13,16 +13,22 @@
 #include <libp2p/basic/scheduler.hpp>
 #include <libp2p/host/host.hpp>
 #include <libp2p/log/sublogger.hpp>
+#include <libp2p/protocol/gossip/time_cache.hpp>
 
 #include "message_cache.hpp"
 #include "message_receiver.hpp"
 #include "peer_set.hpp"
 
+namespace libp2p::protocol::gossip::score {
+  class Score;
+}  // namespace libp2p::protocol::gossip::score
+
 namespace libp2p::protocol::gossip {
 
   class LocalSubscriptions;
   class RemoteSubscriptions;
   class Connectivity;
+  using score::Score;
 
   /// Central component in gossip protocol impl, manages pub-sub logic itself
   class GossipCore : public Gossip,
@@ -71,9 +77,11 @@ namespace libp2p::protocol::gossip {
     void onGraft(const PeerContextPtr &from, const TopicId &topic) override;
     void onPrune(const PeerContextPtr &from,
                  const TopicId &topic,
-                 uint64_t backoff_time) override;
+                 std::optional<std::chrono::seconds> backoff_time) override;
     void onTopicMessage(const PeerContextPtr &from,
                         TopicMessage::Ptr msg) override;
+    void onIDontWant(const PeerContextPtr &from,
+                     const std::vector<MessageId> &message_ids) override;
     void onMessageEnd(const PeerContextPtr &from) override;
 
     /// Periodic heartbeat timer fn
@@ -86,6 +94,7 @@ namespace libp2p::protocol::gossip {
     void onPeerConnection(bool connected, const PeerContextPtr &ctx);
 
     void setTimerHeartbeat();
+    void setTimerScore();
 
     /// Configuration parameters
     const Config config_;
@@ -115,6 +124,14 @@ namespace libp2p::protocol::gossip {
     /// Message cache w/expiration
     MessageCache msg_cache_;
 
+    std::shared_ptr<Score> score_;
+
+    std::shared_ptr<OutboundPeers> outbound_peers_;
+
+    DuplicateCache<MessageId, qtils::BytesStdHash> duplicate_cache_;
+
+    GossipPromises gossip_promises_;
+
     /// Local subscriptions manager (this host subscribed to topics)
     std::shared_ptr<LocalSubscriptions> local_subscriptions_;
 
@@ -132,9 +149,6 @@ namespace libp2p::protocol::gossip {
     /// Network part of gossip component
     std::shared_ptr<Connectivity> connectivity_;
 
-    /// Local {un}subscribe changes to be broadcasted to peers
-    std::map<TopicId, bool> broadcast_on_heartbeat_;
-
     /// Incremented msg sequence number
     uint64_t msg_seq_;
 
@@ -146,6 +160,9 @@ namespace libp2p::protocol::gossip {
 
     /// Logger
     log::SubLogger log_;
+
+    event::Handle new_connection_sub_;
+    event::Handle peer_disconnected_sub_;
   };
 
 }  // namespace libp2p::protocol::gossip
diff --git a/src/protocol/gossip/impl/message_builder.cpp b/src/protocol/gossip/impl/message_builder.cpp
index 3def3b54b..88d002182 100644
--- a/src/protocol/gossip/impl/message_builder.cpp
+++ b/src/protocol/gossip/impl/message_builder.cpp
@@ -7,6 +7,7 @@
 #include "message_builder.hpp"
 
 #include <libp2p/multi/uvarint.hpp>
+#include <qtils/bytestr.hpp>
 
 #include <generated/protocol/gossip/protobuf/rpc.pb.h>
 
@@ -29,6 +30,8 @@ namespace libp2p::protocol::gossip {
     control_pb_msg_->Clear();
     empty_ = true;
     control_not_empty_ = false;
+    subscriptions_.clear();
+    idontwant_.clear();
     ihaves_.clear();
     iwant_.clear();
     messages_added_.clear();
@@ -39,6 +42,8 @@ namespace libp2p::protocol::gossip {
     control_pb_msg_.reset();
     empty_ = true;
     control_not_empty_ = false;
+    std::exchange(subscriptions_, {});
+    std::exchange(idontwant_, {});
     decltype(ihaves_){}.swap(ihaves_);
     decltype(iwant_){}.swap(iwant_);
     decltype(messages_added_){}.swap(messages_added_);
@@ -58,6 +63,19 @@ namespace libp2p::protocol::gossip {
   outcome::result<SharedBuffer> MessageBuilder::serialize() {
     create_protobuf_structures();
 
+    for (auto &[topic, subscribe] : subscriptions_) {
+      auto *subscription = pb_msg_->add_subscriptions();
+      subscription->set_subscribe(subscribe);
+      subscription->set_topicid(topic);
+    }
+
+    if (not idontwant_.empty()) {
+      auto *idontwant = control_pb_msg_->add_idontwant();
+      for (auto &msg_id : idontwant_) {
+        *idontwant->add_message_ids() = qtils::byte2str(msg_id);
+      }
+    }
+
     for (auto &[topic, message_ids] : ihaves_) {
       auto *ih = control_pb_msg_->add_ihave();
       ih->set_topicid(topic);
@@ -109,11 +127,16 @@ namespace libp2p::protocol::gossip {
   }
 
   void MessageBuilder::addSubscription(bool subscribe, const TopicId &topic) {
-    create_protobuf_structures();
+    auto it = subscriptions_.emplace(topic, subscribe).first;
+    if (it->second != subscribe) {
+      subscriptions_.erase(it);
+    }
+    empty_ = false;
+  }
 
-    auto *dst = pb_msg_->add_subscriptions();
-    dst->set_subscribe(subscribe);
-    dst->set_topicid(topic);
+  void MessageBuilder::addIDontWant(const MessageId &msg_id) {
+    idontwant_.emplace(msg_id);
+    control_not_empty_ = true;
     empty_ = false;
   }
 
@@ -137,10 +160,15 @@ namespace libp2p::protocol::gossip {
     empty_ = false;
   }
 
-  void MessageBuilder::addPrune(const TopicId &topic) {
+  void MessageBuilder::addPrune(const TopicId &topic,
+                                std::optional<std::chrono::seconds> backoff) {
     create_protobuf_structures();
 
-    control_pb_msg_->add_prune()->set_topicid(topic);
+    auto *prune = control_pb_msg_->add_prune();
+    prune->set_topicid(topic);
+    if (backoff.has_value()) {
+      prune->set_backoff(backoff->count());
+    }
     control_not_empty_ = true;
     empty_ = false;
   }
@@ -155,18 +183,7 @@ namespace libp2p::protocol::gossip {
     }
     messages_added_.insert(msg_id);
 
-    auto *dst = pb_msg_->add_publish();
-    dst->set_from(msg.from.data(), msg.from.size());
-    dst->set_data(msg.data.data(), msg.data.size());
-    dst->set_seqno(msg.seq_no.data(), msg.seq_no.size());
-    dst->set_topic(msg.topic);
-    if (msg.signature) {
-      dst->set_signature(msg.signature.value().data(),
-                         msg.signature.value().size());
-    }
-    if (msg.key) {
-      dst->set_key(msg.key.value().data(), msg.key.value().size());
-    }
+    toPb(*pb_msg_->add_publish(), msg);
     empty_ = false;
   }
 
@@ -188,4 +205,26 @@ namespace libp2p::protocol::gossip {
     }
     return signable;
   }
+
+  void MessageBuilder::toPb(pubsub::pb::Message &pb_message,
+                            const TopicMessage &message) {
+    pb_message.set_from(message.from.data(), message.from.size());
+    pb_message.set_data(message.data.data(), message.data.size());
+    pb_message.set_seqno(message.seq_no.data(), message.seq_no.size());
+    pb_message.set_topic(message.topic);
+    if (message.signature) {
+      pb_message.set_signature(message.signature.value().data(),
+                               message.signature.value().size());
+    }
+    if (message.key) {
+      pb_message.set_key(message.key.value().data(),
+                         message.key.value().size());
+    }
+  }
+
+  size_t MessageBuilder::pbSize(const TopicMessage &message) {
+    static thread_local pubsub::pb::Message pb_message;
+    toPb(pb_message, message);
+    return pb_message.ByteSizeLong();
+  }
 }  // namespace libp2p::protocol::gossip
diff --git a/src/protocol/gossip/impl/message_builder.hpp b/src/protocol/gossip/impl/message_builder.hpp
index e0e2585da..df529daa8 100644
--- a/src/protocol/gossip/impl/message_builder.hpp
+++ b/src/protocol/gossip/impl/message_builder.hpp
@@ -7,12 +7,14 @@
 #pragma once
 
 #include <map>
+#include <unordered_map>
 #include <unordered_set>
 
 #include "common.hpp"
 
 namespace pubsub::pb {
   // protobuf entities forward declaration
+  class Message;
   class RPC;
   class ControlMessage;
 }  // namespace pubsub::pb
@@ -44,6 +46,8 @@ namespace libp2p::protocol::gossip {
     /// Adds subscription notification
     void addSubscription(bool subscribe, const TopicId &topic);
 
+    void addIDontWant(const MessageId &msg_id);
+
     /// Adds "I have" notification
     void addIHave(const TopicId &topic, const MessageId &msg_id);
 
@@ -54,13 +58,18 @@ namespace libp2p::protocol::gossip {
     void addGraft(const TopicId &topic);
 
     /// Adds prune request
-    void addPrune(const TopicId &topic);
+    void addPrune(const TopicId &topic,
+                  std::optional<std::chrono::seconds> backoff);
 
     /// Adds message to be forwarded
     void addMessage(const TopicMessage &msg, const MessageId &msg_id);
 
     static outcome::result<Bytes> signableMessage(const TopicMessage &msg);
 
+    static void toPb(pubsub::pb::Message &pb_message,
+                     const TopicMessage &message);
+    static size_t pbSize(const TopicMessage &message);
+
    private:
     /// Creates protobuf structures if needed
     void create_protobuf_structures();
@@ -73,6 +82,8 @@ namespace libp2p::protocol::gossip {
     std::unique_ptr<pubsub::pb::ControlMessage> control_pb_msg_;
     bool empty_;
     bool control_not_empty_;
+    std::unordered_map<TopicId, bool> subscriptions_;
+    std::unordered_set<MessageId> idontwant_;
 
     /// Intermediate struct for building IHave messages
     std::map<TopicId, std::vector<MessageId>> ihaves_;
diff --git a/src/protocol/gossip/impl/message_parser.cpp b/src/protocol/gossip/impl/message_parser.cpp
index 15fc22924..9a845247b 100644
--- a/src/protocol/gossip/impl/message_parser.cpp
+++ b/src/protocol/gossip/impl/message_parser.cpp
@@ -7,6 +7,7 @@
 #include "message_parser.hpp"
 
 #include <libp2p/log/logger.hpp>
+#include <qtils/bytestr.hpp>
 
 #include "message_receiver.hpp"
 
@@ -52,6 +53,15 @@ namespace libp2p::protocol::gossip {
     if (pb_msg_->has_control()) {
       const auto &c = pb_msg_->control();
 
+      std::vector<MessageId> idontwant_message_ids;
+      for (const auto &idontwant : c.idontwant()) {
+        for (auto &msg_id : idontwant.message_ids()) {
+          idontwant_message_ids.emplace_back(
+              qtils::asVec(qtils::str2byte(msg_id)));
+        }
+      }
+      receiver.onIDontWant(from, idontwant_message_ids);
+
       for (const auto &h : c.ihave()) {
         if (!h.has_topicid() || h.messageids_size() == 0) {
           continue;
@@ -88,12 +98,10 @@ namespace libp2p::protocol::gossip {
         if (!pr.has_topicid()) {
           continue;
         }
-        uint64_t backoff_time = 60;
+        std::optional<std::chrono::seconds> backoff_time;
         if (pr.has_backoff()) {
-          backoff_time = pr.backoff();
+          backoff_time = std::chrono::seconds{pr.backoff()};
         }
-        log()->debug(
-            "prune backoff={}, {} peers", backoff_time, pr.peers_size());
         for (const auto &peer : pr.peers()) {
           // TODO(artem): meshsub 1.1.0 + signed peer records NYI
 
diff --git a/src/protocol/gossip/impl/message_receiver.hpp b/src/protocol/gossip/impl/message_receiver.hpp
index 22401a7eb..153c3111c 100644
--- a/src/protocol/gossip/impl/message_receiver.hpp
+++ b/src/protocol/gossip/impl/message_receiver.hpp
@@ -37,12 +37,15 @@ namespace libp2p::protocol::gossip {
     /// backoff_time seconds
     virtual void onPrune(const PeerContextPtr &from,
                          const TopicId &topic,
-                         uint64_t backoff_time) = 0;
+                         std::optional<std::chrono::seconds> backoff_time) = 0;
 
     /// Message received
     virtual void onTopicMessage(const PeerContextPtr &from,
                                 TopicMessage::Ptr msg) = 0;
 
+    virtual void onIDontWant(const PeerContextPtr &from,
+                             const std::vector<MessageId> &message_ids) = 0;
+
     /// Current wire protocol message dispatch ended
     virtual void onMessageEnd(const PeerContextPtr &from) = 0;
   };
diff --git a/src/protocol/gossip/impl/peer_context.cpp b/src/protocol/gossip/impl/peer_context.cpp
index b3fa791c8..163633c57 100644
--- a/src/protocol/gossip/impl/peer_context.cpp
+++ b/src/protocol/gossip/impl/peer_context.cpp
@@ -22,6 +22,22 @@ namespace libp2p::protocol::gossip {
         str(makeStringRepr(peer_id)),
         message_builder(std::make_shared<MessageBuilder>()) {}
 
+  bool PeerContext::isFloodsub() const {
+    return not peer_kind or peer_kind.value() == PeerKind::Floodsub;
+  }
+
+  bool PeerContext::isGossipsub() const {
+    return peer_kind and peer_kind.value() >= PeerKind::Gossipsub;
+  }
+
+  bool PeerContext::isGossipsubv1_1() const {
+    return peer_kind and peer_kind.value() >= PeerKind::Gossipsubv1_1;
+  }
+
+  bool PeerContext::isGossipsubv1_2() const {
+    return peer_kind and peer_kind.value() >= PeerKind::Gossipsubv1_2;
+  }
+
   bool operator<(const PeerContextPtr &ctx, const peer::PeerId &peer) {
     if (!ctx) {
       return false;
diff --git a/src/protocol/gossip/impl/peer_context.hpp b/src/protocol/gossip/impl/peer_context.hpp
index 56f33538c..2f06537c4 100644
--- a/src/protocol/gossip/impl/peer_context.hpp
+++ b/src/protocol/gossip/impl/peer_context.hpp
@@ -7,6 +7,8 @@
 #pragma once
 
 #include <libp2p/common/metrics/instance_count.hpp>
+#include <libp2p/protocol/gossip/time_cache.hpp>
+#include <optional>
 
 #include "common.hpp"
 
@@ -42,6 +44,10 @@ namespace libp2p::protocol::gossip {
     /// If true, then outbound connection is in progress
     bool is_connecting = false;
 
+    std::optional<PeerKind> peer_kind;
+
+    IDontWantCache<MessageId, qtils::BytesStdHash> idontwant;
+
     ~PeerContext() = default;
     PeerContext(PeerContext &&) = delete;
     PeerContext(const PeerContext &) = delete;
@@ -50,6 +56,11 @@ namespace libp2p::protocol::gossip {
 
     explicit PeerContext(peer::PeerId id);
 
+    bool isFloodsub() const;
+    bool isGossipsub() const;
+    bool isGossipsubv1_1() const;
+    bool isGossipsubv1_2() const;
+
     LIBP2P_METRICS_INSTANCE_COUNT_IF_ENABLED(
         libp2p::protocol::gossip::PeerContext);
   };
diff --git a/src/protocol/gossip/impl/peer_set.cpp b/src/protocol/gossip/impl/peer_set.cpp
index fa9d13410..57e29a2b6 100644
--- a/src/protocol/gossip/impl/peer_set.cpp
+++ b/src/protocol/gossip/impl/peer_set.cpp
@@ -24,7 +24,11 @@ namespace libp2p::protocol::gossip {
   }
 
   bool PeerSet::contains(const peer::PeerId &id) const {
-    return peers_.count(id) != 0;
+    return peers_.contains(id);
+  }
+
+  bool PeerSet::contains(const PeerContextPtr &ctx) const {
+    return peers_.contains(ctx);
   }
 
   bool PeerSet::insert(PeerContextPtr ctx) {
diff --git a/src/protocol/gossip/impl/peer_set.hpp b/src/protocol/gossip/impl/peer_set.hpp
index c9e08c660..c326617aa 100644
--- a/src/protocol/gossip/impl/peer_set.hpp
+++ b/src/protocol/gossip/impl/peer_set.hpp
@@ -21,6 +21,7 @@ namespace libp2p::protocol::gossip {
 
     /// Returns if the set contains such a peer
     bool contains(const peer::PeerId &id) const;
+    bool contains(const PeerContextPtr &ctx) const;
 
     /// Inserts peer context into set, returns false if already inserted
     bool insert(PeerContextPtr ctx);
@@ -56,6 +57,18 @@ namespace libp2p::protocol::gossip {
     /// Conditionally erases peers from the set
     void eraseIf(const FilterCallback &filter);
 
+    auto begin() const {
+      return peers_.begin();
+    }
+    auto end() const {
+      return peers_.end();
+    }
+    void extend(auto &&peers) {
+      for (auto &ctx : peers) {
+        insert(ctx);
+      }
+    }
+
    private:
     std::set<PeerContextPtr, std::less<>> peers_;
   };
diff --git a/src/protocol/gossip/impl/remote_subscriptions.cpp b/src/protocol/gossip/impl/remote_subscriptions.cpp
index 4a2f5e7aa..432c39cbf 100644
--- a/src/protocol/gossip/impl/remote_subscriptions.cpp
+++ b/src/protocol/gossip/impl/remote_subscriptions.cpp
@@ -7,31 +7,47 @@
 #include "remote_subscriptions.hpp"
 
 #include <algorithm>
+#include <libp2p/protocol/gossip/explicit_peers.hpp>
 
+#include "choose_peers.hpp"
 #include "connectivity.hpp"
 #include "message_builder.hpp"
 
 namespace libp2p::protocol::gossip {
 
-  RemoteSubscriptions::RemoteSubscriptions(const Config &config,
-                                           Connectivity &connectivity,
-                                           basic::Scheduler &scheduler,
-                                           log::SubLogger &log)
+  RemoteSubscriptions::RemoteSubscriptions(
+      const Config &config,
+      Connectivity &connectivity,
+      std::shared_ptr<Score> score,
+      std::shared_ptr<OutboundPeers> outbound_peers,
+      std::shared_ptr<basic::Scheduler> scheduler,
+      log::SubLogger &log)
       : config_(config),
         connectivity_(connectivity),
-        scheduler_(scheduler),
+        choose_peers_{std::make_shared<ChoosePeers>()},
+        explicit_peers_{std::make_shared<ExplicitPeers>()},
+        score_{std::move(score)},
+        outbound_peers_{std::move(outbound_peers)},
+        scheduler_{std::move(scheduler)},
         log_(log) {}
 
   void RemoteSubscriptions::onSelfSubscribed(bool subscribed,
                                              const TopicId &topic) {
+    if (subscribed == isSubscribed(topic)) {
+      return;
+    }
     auto res = getItem(topic, subscribed);
     if (!res) {
       log_.error("error in self subscribe to {}", topic);
       return;
     }
     TopicSubscriptions &subs = res.value();
-    subs.onSelfSubscribed(subscribed);
-    if (subs.empty()) {
+    if (subscribed) {
+      subs.subscribe();
+    } else {
+      subs.unsubscribe();
+    }
+    if (not subs.isUsed()) {
       table_.erase(topic);
     }
     log_.debug("self {} {}",
@@ -48,7 +64,7 @@ namespace libp2p::protocol::gossip {
     }
     log_.debug("peer {} subscribing to {}", peer->str, topic);
 
-    auto res = getItem(topic, true);
+    auto res = getItem(topic, false);
     if (!res) {
       // not error in this case, this is request from wire...
       log_.debug("entry doesnt exist for {}", topic);
@@ -78,7 +94,7 @@ namespace libp2p::protocol::gossip {
     TopicSubscriptions &subs = res.value();
 
     subs.onPeerUnsubscribed(peer);
-    if (subs.empty()) {
+    if (not subs.isUsed()) {
       table_.erase(topic);
     }
   }
@@ -92,38 +108,44 @@ namespace libp2p::protocol::gossip {
     }
   }
 
-  bool RemoteSubscriptions::hasTopic(const TopicId &topic) const {
-    return table_.count(topic) != 0;
+  bool RemoteSubscriptions::isSubscribed(const TopicId &topic) const {
+    auto it = table_.find(topic);
+    return it != table_.end() and it->second.isSubscribed();
   }
 
   void RemoteSubscriptions::onGraft(const PeerContextPtr &peer,
                                     const TopicId &topic) {
+    // implicit subscribe on graft
+    peer->subscribed_to.insert(topic);
     auto res = getItem(topic, false);
     if (!res) {
       // we don't have this topic anymore
-      peer->message_builder->addPrune(topic);
-      connectivity_.peerIsWritable(peer, true);
+      peer->message_builder->addPrune(
+          topic,
+          peer->isGossipsubv1_1() ? std::make_optional(config_.prune_backoff)
+                                  : std::nullopt);
+      connectivity_.peerIsWritable(peer);
       return;
     }
     res.value().onGraft(peer);
   }
 
-  void RemoteSubscriptions::onPrune(const PeerContextPtr &peer,
-                                    const TopicId &topic,
-                                    uint64_t backoff_time) {
+  void RemoteSubscriptions::onPrune(
+      const PeerContextPtr &peer,
+      const TopicId &topic,
+      std::optional<std::chrono::seconds> backoff_time) {
     auto res = getItem(topic, false);
     if (!res) {
       return;
     }
-    res.value().onPrune(peer,
-                        scheduler_.now() + std::chrono::seconds(backoff_time));
+    res.value().onPrune(peer, backoff_time);
   }
 
   void RemoteSubscriptions::onNewMessage(
       const boost::optional<PeerContextPtr> &from,
       const TopicMessage::Ptr &msg,
       const MessageId &msg_id) {
-    auto now = scheduler_.now();
+    auto now = scheduler_->now();
     bool is_published_locally = !from.has_value();
     auto res = getItem(msg->topic, is_published_locally);
     if (!res) {
@@ -134,10 +156,10 @@ namespace libp2p::protocol::gossip {
   }
 
   void RemoteSubscriptions::onHeartbeat() {
-    auto now = scheduler_.now();
+    auto now = scheduler_->now();
     for (auto it = table_.begin(); it != table_.end();) {
       it->second.onHeartbeat(now);
-      if (it->second.empty()) {
+      if (not it->second.isUsed()) {
         // fanout interval expired - clean up
         log_.debug("deleted entry for topic {}", it->first);
         it = table_.erase(it);
@@ -154,14 +176,17 @@ namespace libp2p::protocol::gossip {
       return it->second;
     }
     if (create_if_not_exist) {
-      auto [it, _] = table_.emplace(
-          topic, TopicSubscriptions(topic, config_, connectivity_, log_));
+      auto [it, _] = table_.emplace(topic,
+                                    TopicSubscriptions(topic,
+                                                       config_,
+                                                       connectivity_,
+                                                       scheduler_,
+                                                       choose_peers_,
+                                                       explicit_peers_,
+                                                       score_,
+                                                       outbound_peers_,
+                                                       log_));
       TopicSubscriptions &item = it->second;
-      connectivity_.getConnectedPeers().selectIf(
-          [&item](const PeerContextPtr &ctx) { item.onPeerSubscribed(ctx); },
-          [&topic](const PeerContextPtr &ctx) {
-            return ctx->subscribed_to.count(topic) != 0;
-          });
       log_.debug("created entry for topic {}", topic);
       return item;
     }
diff --git a/src/protocol/gossip/impl/remote_subscriptions.hpp b/src/protocol/gossip/impl/remote_subscriptions.hpp
index 765d5bcdb..8eec4fc6f 100644
--- a/src/protocol/gossip/impl/remote_subscriptions.hpp
+++ b/src/protocol/gossip/impl/remote_subscriptions.hpp
@@ -20,7 +20,9 @@ namespace libp2p::protocol::gossip {
     /// GossipCore and lives only within its scope
     RemoteSubscriptions(const Config &config,
                         Connectivity &connectivity,
-                        basic::Scheduler &scheduler,
+                        std::shared_ptr<Score> score,
+                        std::shared_ptr<OutboundPeers> outbound_peers,
+                        std::shared_ptr<basic::Scheduler> scheduler,
                         log::SubLogger &log);
 
     /// This host subscribes or unsubscribes
@@ -38,7 +40,7 @@ namespace libp2p::protocol::gossip {
     void onPeerDisconnected(const PeerContextPtr &peer);
 
     /// Returns if topic exists in the table
-    bool hasTopic(const TopicId &topic) const;
+    bool isSubscribed(const TopicId &topic) const;
 
     /// Remote peer adds topic into its mesh
     void onGraft(const PeerContextPtr &peer, const TopicId &topic);
@@ -46,7 +48,7 @@ namespace libp2p::protocol::gossip {
     /// Remote peer removes topic from its mesh
     void onPrune(const PeerContextPtr &peer,
                  const TopicId &topic,
-                 uint64_t backoff_time);
+                 std::optional<std::chrono::seconds> backoff_time);
 
     /// Forwards message to its topics. If 'from' is not set then the message is
     /// published locally
@@ -64,10 +66,12 @@ namespace libp2p::protocol::gossip {
 
     const Config &config_;
     Connectivity &connectivity_;
-    basic::Scheduler &scheduler_;
+    std::shared_ptr<ChoosePeers> choose_peers_;
+    std::shared_ptr<ExplicitPeers> explicit_peers_;
+    std::shared_ptr<Score> score_;
+    std::shared_ptr<OutboundPeers> outbound_peers_;
+    std::shared_ptr<basic::Scheduler> scheduler_;
 
-    // TODO(artem): bound table size (which may grow!)
-    // by removing items not subscribed to locally. LRU(???)
     std::unordered_map<TopicId, TopicSubscriptions> table_;
 
     log::SubLogger &log_;
diff --git a/src/protocol/gossip/impl/stream.cpp b/src/protocol/gossip/impl/stream.cpp
index 98baf3a74..d8f4eace5 100644
--- a/src/protocol/gossip/impl/stream.cpp
+++ b/src/protocol/gossip/impl/stream.cpp
@@ -23,7 +23,7 @@ namespace libp2p::protocol::gossip {
                  const Config &config,
                  basic::Scheduler &scheduler,
                  const Feedback &feedback,
-                 MessageReceiver &msg_receiver,
+                 std::weak_ptr<MessageReceiver> msg_receiver,
                  std::shared_ptr<connection::Stream> stream,
                  PeerContextPtr peer)
       : stream_id_(stream_id),
@@ -31,7 +31,7 @@ namespace libp2p::protocol::gossip {
         scheduler_(scheduler),
         max_message_size_(config.max_message_size),
         feedback_(feedback),
-        msg_receiver_(msg_receiver),
+        msg_receiver_{std::move(msg_receiver)},
         stream_(std::move(stream)),
         peer_(std::move(peer)),
         read_buffer_(std::make_shared<std::vector<uint8_t>>()) {
@@ -119,7 +119,12 @@ namespace libp2p::protocol::gossip {
       return;
     }
 
-    parser.dispatch(peer_, msg_receiver_);
+    if (auto msg_receiver = msg_receiver_.lock()) {
+      parser.dispatch(peer_, *msg_receiver);
+    } else {
+      close();
+      return;
+    }
 
     // reads again
     read();
diff --git a/src/protocol/gossip/impl/stream.hpp b/src/protocol/gossip/impl/stream.hpp
index 12477a061..ff3a9790c 100644
--- a/src/protocol/gossip/impl/stream.hpp
+++ b/src/protocol/gossip/impl/stream.hpp
@@ -34,7 +34,7 @@ namespace libp2p::protocol::gossip {
            const Config &config,
            basic::Scheduler &scheduler,
            const Feedback &feedback,
-           MessageReceiver &msg_receiver,
+           std::weak_ptr<MessageReceiver> msg_receiver,
            std::shared_ptr<connection::Stream> stream,
            PeerContextPtr peer);
 
@@ -61,7 +61,7 @@ namespace libp2p::protocol::gossip {
     basic::Scheduler &scheduler_;
     const size_t max_message_size_;
     const Feedback &feedback_;
-    MessageReceiver &msg_receiver_;
+    std::weak_ptr<MessageReceiver> msg_receiver_;
     std::shared_ptr<connection::Stream> stream_;
     PeerContextPtr peer_;
 
diff --git a/src/protocol/gossip/impl/topic_subscriptions.cpp b/src/protocol/gossip/impl/topic_subscriptions.cpp
index 37e8d74b0..ecb1f4190 100644
--- a/src/protocol/gossip/impl/topic_subscriptions.cpp
+++ b/src/protocol/gossip/impl/topic_subscriptions.cpp
@@ -8,7 +8,10 @@
 
 #include <algorithm>
 #include <cassert>
+#include <libp2p/protocol/gossip/explicit_peers.hpp>
+#include <libp2p/protocol/gossip/score.hpp>
 
+#include "choose_peers.hpp"
 #include "connectivity.hpp"
 #include "message_builder.hpp"
 
@@ -29,21 +32,43 @@ namespace libp2p::protocol::gossip {
 
   }  // namespace
 
-  TopicSubscriptions::TopicSubscriptions(TopicId topic,
-                                         const Config &config,
-                                         Connectivity &connectivity,
-                                         log::SubLogger &log)
+  TopicSubscriptions::TopicSubscriptions(
+      TopicId topic,
+      const Config &config,
+      Connectivity &connectivity,
+      std::shared_ptr<basic::Scheduler> scheduler,
+      std::shared_ptr<ChoosePeers> choose_peers,
+      std::shared_ptr<ExplicitPeers> explicit_peers,
+      std::shared_ptr<Score> score,
+      std::shared_ptr<OutboundPeers> outbound_peers,
+      log::SubLogger &log)
       : topic_(std::move(topic)),
         config_(config),
         connectivity_(connectivity),
+        scheduler_{std::move(scheduler)},
+        choose_peers_{std::move(choose_peers)},
+        explicit_peers_{std::move(explicit_peers)},
+        score_{std::move(score)},
+        outbound_peers_{std::move(outbound_peers)},
         self_subscribed_(false),
-        fanout_period_ends_(0),
-        log_(log) {}
+        log_(log) {
+    connectivity_.getConnectedPeers().selectIf(
+        [&](const PeerContextPtr &ctx) { subscribed_peers_.insert(ctx); },
+        [&](const PeerContextPtr &ctx) {
+          return ctx->subscribed_to.count(topic_) != 0;
+        });
+    if (config_.history_gossip == 0) {
+      throw std::logic_error{"gossip config.history_gossip must not be zero"};
+    }
+    history_gossip_.resize(config_.history_gossip);
+  }
 
-  bool TopicSubscriptions::empty() const {
-    return (not self_subscribed_)
-        && (fanout_period_ends_ == std::chrono::milliseconds::zero())
-        && subscribed_peers_.empty() && mesh_peers_.empty();
+  bool TopicSubscriptions::isUsed() const {
+    return self_subscribed_ or fanout_.has_value();
+  }
+
+  bool TopicSubscriptions::isSubscribed() const {
+    return self_subscribed_;
   }
 
   void TopicSubscriptions::onNewMessage(
@@ -53,199 +78,434 @@ namespace libp2p::protocol::gossip {
       Time now) {
     bool is_published_locally = !from.has_value();
 
-    if (is_published_locally) {
-      fanout_period_ends_ = now + config_.seen_cache_lifetime_msec;
-      log_.debug("setting fanout period for {}, {}->{}",
-                 topic_,
-                 now.count(),
-                 fanout_period_ends_.count());
+    if (not self_subscribed_ and not is_published_locally) {
+      return;
     }
 
     auto origin = peerFrom(*msg);
 
-    mesh_peers_.selectAll(
-        [this, &msg, &msg_id, &from, &origin](const PeerContextPtr &ctx) {
-          assert(ctx->message_builder);
-
-          if (needToForward(ctx, from, origin)) {
-            ctx->message_builder->addMessage(*msg, msg_id);
-
-            // forward immediately to those in mesh
-            connectivity_.peerIsWritable(ctx, true);
-          }
-        });
-
-    auto peers = subscribed_peers_.selectRandomPeers(config_.D_max * 2);
-    for (const auto &ctx : peers) {
-      assert(ctx->message_builder);
-
-      if (needToForward(ctx, from, origin)) {
-        ctx->message_builder->addIHave(topic_, msg_id);
+    auto add_peer = [&](const PeerContextPtr &ctx) {
+      if (needToForward(ctx, from, origin)
+          and not ctx->idontwant.contains(msg_id)) {
+        ctx->message_builder->addMessage(*msg, msg_id);
+        connectivity_.peerIsWritable(ctx);
+      }
+    };
+
+    if (config_.flood_publish and is_published_locally) {
+      for (auto &ctx : subscribed_peers_) {
+        if (explicit_peers_->contains(ctx->peer_id)
+            or not score_->below(ctx->peer_id,
+                                 config_.score.publish_threshold)) {
+          add_peer(ctx);
+        }
+      }
+    } else {
+      for (auto &ctx : subscribed_peers_) {
+        if (ctx->isFloodsub()
+            and not score_->below(ctx->peer_id,
+                                  config_.score.publish_threshold)) {
+          add_peer(ctx);
+        }
+      }
 
-        // local messages announce themselves immediately
-        connectivity_.peerIsWritable(ctx, is_published_locally);
+      if (self_subscribed_) {
+        for (auto &ctx : mesh_peers_) {
+          add_peer(ctx);
+        }
+      } else {
+        if (not fanout_.has_value()) {
+          fanout_.emplace();
+        }
+        if (fanout_->peers.empty()) {
+          fanout_->peers.extend(choose_peers_->choose(
+              subscribed_peers_,
+              [&](const PeerContextPtr &ctx) {
+                return not explicit_peers_->contains(ctx->peer_id)
+                   and not score_->below(ctx->peer_id,
+                                         config_.score.publish_threshold);
+              },
+              config_.D));
+        }
+        fanout_->until = now + config_.fanout_ttl;
+        for (auto &ctx : fanout_->peers) {
+          add_peer(ctx);
+        }
       }
     }
 
-    seen_cache_.emplace_back(now + config_.seen_cache_lifetime_msec, msg_id);
+    history_gossip_.back().emplace_back(msg_id);
 
-    log_.debug("message forwarded, topic={}, m={}, s={}",
-               topic_,
-               mesh_peers_.size(),
-               subscribed_peers_.size());
+    if ((not is_published_locally or config_.idontwant_on_publish)
+        and MessageBuilder::pbSize(*msg)
+                > config_.idontwant_message_size_threshold) {
+      for (auto &ctx : mesh_peers_) {
+        if (ctx->isGossipsubv1_2()) {
+          ctx->message_builder->addIDontWant(msg_id);
+          connectivity_.peerIsWritable(ctx);
+        }
+      }
+    }
   }
 
   void TopicSubscriptions::onHeartbeat(Time now) {
-    if (self_subscribed_ && !subscribed_peers_.empty()) {
+    ++heartbeat_ticks_;
+    auto slack = config_.backoff_slack * config_.heartbeat_interval_msec;
+    for (auto it = dont_bother_until_.begin();
+         it != dont_bother_until_.end();) {
+      if (it->second + slack > now) {
+        ++it;
+      } else {
+        it = dont_bother_until_.erase(it);
+      }
+    }
+
+    if (self_subscribed_) {
       // add/remove mesh members according to desired network density D
-      size_t sz = mesh_peers_.size();
-
-      if (sz < config_.D_min) {
-        auto peers = subscribed_peers_.selectRandomPeers(config_.D_min - sz);
-        for (auto &p : peers) {
-          auto it = dont_bother_until_.find(p);
-          if (it != dont_bother_until_.end()) {
-            if (it->second < now) {
-              dont_bother_until_.erase(it);
-            } else {
+      mesh_peers_.eraseIf([&](const PeerContextPtr &ctx) {
+        if (score_->below(ctx->peer_id, config_.score.zero)) {
+          sendPrune(ctx, false);
+          return true;
+        }
+        return false;
+      });
+      auto outbound = static_cast<size_t>(
+          std::ranges::count_if(mesh_peers_, [&](const PeerContextPtr &ctx) {
+            return outbound_peers_->contains(ctx->peer_id);
+          }));
+      if (mesh_peers_.size() < config_.D_min) {
+        for (auto &ctx : choose_peers_->choose(
+                 subscribed_peers_,
+                 [&](const PeerContextPtr &ctx) {
+                   return not mesh_peers_.contains(ctx)
+                      and not explicit_peers_->contains(ctx->peer_id)
+                      and not isBackoffWithSlack(ctx->peer_id)
+                      and not score_->below(ctx->peer_id, config_.score.zero);
+                 },
+                 config_.D - mesh_peers_.size())) {
+          if (outbound_peers_->contains(ctx->peer_id)) {
+            ++outbound;
+          }
+          addToMesh(ctx);
+        }
+      } else if (mesh_peers_.size() > config_.D_max) {
+        std::vector<PeerId> shuffled;
+        for (auto &ctx : mesh_peers_) {
+          shuffled.emplace_back(ctx->peer_id);
+        }
+        choose_peers_->shuffle(shuffled);
+        std::ranges::sort(shuffled, [&](const PeerId &l, const PeerId &r) {
+          return score_->score(r) < score_->score(l);
+        });
+        if (shuffled.size() > config_.retain_scores) {
+          choose_peers_->shuffle(
+              std::span{shuffled}.subspan(config_.retain_scores));
+        }
+        for (auto &peer_id : shuffled) {
+          if (mesh_peers_.size() <= config_.D_max) {
+            break;
+          }
+          if (outbound_peers_->contains(peer_id)) {
+            if (outbound <= config_.mesh_outbound_min) {
               continue;
             }
+            --outbound;
           }
-
-          addToMesh(p);
-          subscribed_peers_.erase(p->peer_id);
+          auto ctx = mesh_peers_.find(peer_id).value();
+          sendPrune(ctx, false);
+          mesh_peers_.erase(ctx->peer_id);
         }
-      } else if (sz > config_.D_max) {
-        auto peers = mesh_peers_.selectRandomPeers(sz - config_.D_max);
-        for (auto &p : peers) {
-          removeFromMesh(p);
-          mesh_peers_.erase(p->peer_id);
+      }
+      if (mesh_peers_.size() >= config_.D_min) {
+        if (outbound < config_.mesh_outbound_min) {
+          for (auto &ctx : choose_peers_->choose(
+                   subscribed_peers_,
+                   [&](const PeerContextPtr &ctx) {
+                     return not mesh_peers_.contains(ctx)
+                        and not explicit_peers_->contains(ctx->peer_id)
+                        and not isBackoffWithSlack(ctx->peer_id)
+                        and not score_->below(ctx->peer_id, config_.score.zero)
+                        and outbound_peers_->contains(ctx->peer_id);
+                   },
+                   config_.mesh_outbound_min - outbound)) {
+            addToMesh(ctx);
+          }
+        }
+      }
+      if (heartbeat_ticks_ % config_.opportunistic_graft_ticks == 0
+          and mesh_peers_.size() > 1) {
+        std::vector<double> scores;
+        scores.reserve(mesh_peers_.size());
+        for (auto &ctx : mesh_peers_) {
+          scores.emplace_back(score_->score(ctx->peer_id));
+        }
+        std::ranges::sort(scores);
+        auto middle = scores.size() / 2;
+        auto median = scores.size() % 2 == 0
+                        ? (scores[middle - 1] + scores[middle + 1]) / 2
+                        : scores[middle];
+        if (median < config_.score.opportunistic_graft_threshold) {
+          for (auto &ctx : choose_peers_->choose(
+                   subscribed_peers_,
+                   [&](const PeerContextPtr &ctx) {
+                     return not mesh_peers_.contains(ctx)
+                        and not explicit_peers_->contains(ctx->peer_id)
+                        and not isBackoffWithSlack(ctx->peer_id)
+                        and score_->score(ctx->peer_id) > median;
+                   },
+                   config_.opportunistic_graft_peers)) {
+            addToMesh(ctx);
+          }
         }
       }
     }
 
     // fanout ends some time after this host ends publishing to the topic,
     // to save space and traffic
-    if (fanout_period_ends_ != Time::zero() && fanout_period_ends_ < now) {
-      fanout_period_ends_ = Time::zero();
+    if (fanout_.has_value() and fanout_->until < now) {
+      fanout_.reset();
       log_.debug("fanout period reset for {}", topic_);
     }
+    if (fanout_.has_value()) {
+      fanout_->peers.eraseIf([&](const PeerContextPtr &ctx) {
+        return score_->below(ctx->peer_id, config_.score.publish_threshold);
+      });
+      if (fanout_->peers.size() < config_.D) {
+        fanout_->peers.extend(choose_peers_->choose(
+            subscribed_peers_,
+            [&](const PeerContextPtr &ctx) {
+              return not fanout_->peers.contains(ctx)
+                 and not explicit_peers_->contains(ctx->peer_id)
+                 and not score_->below(ctx->peer_id,
+                                       config_.score.publish_threshold);
+            },
+            config_.D - fanout_->peers.size()));
+      }
+    }
+
+    emitGossip();
 
     // shift msg ids cache
-    auto seen_cache_size = seen_cache_.size();
-    bool changed = false;
+    history_gossip_.pop_front();
+    history_gossip_.emplace_back();
+  }
 
-    if (seen_cache_size > config_.seen_cache_limit) {
-      auto b = seen_cache_.begin();
-      auto e = b + ssize_t(seen_cache_size - config_.seen_cache_limit);
-      seen_cache_.erase(b, e);
-      changed = true;
-    } else if (seen_cache_size != 0) {
-      auto it = std::find_if(seen_cache_.begin(),
-                             seen_cache_.end(),
-                             [now](const auto &p) { return p.first >= now; });
-      if (it != seen_cache_.begin()) {
-        seen_cache_.erase(seen_cache_.begin(), it);
-        changed = true;
+  void TopicSubscriptions::subscribe() {
+    if (self_subscribed_) {
+      return;
+    }
+    self_subscribed_ = true;
+    if (fanout_.has_value()) {
+      for (auto &ctx : fanout_->peers) {
+        if (explicit_peers_->contains(ctx->peer_id)) {
+          continue;
+        }
+        if (score_->below(ctx->peer_id, config_.score.publish_threshold)) {
+          continue;
+        }
+        if (isBackoffWithSlack(ctx->peer_id)) {
+          continue;
+        }
+        if (mesh_peers_.size() >= config_.D) {
+          break;
+        }
+        addToMesh(ctx);
       }
+      fanout_.reset();
     }
-
-    if (changed) {
-      log_.debug("seen cache size changed {}->{} for {}",
-                 seen_cache_size,
-                 seen_cache_.size(),
-                 topic_);
+    if (mesh_peers_.size() < config_.D) {
+      for (auto &ctx : choose_peers_->choose(
+               subscribed_peers_,
+               [&](const PeerContextPtr &ctx) {
+                 return not mesh_peers_.contains(ctx)
+                    and not explicit_peers_->contains(ctx->peer_id)
+                    and not isBackoffWithSlack(ctx->peer_id)
+                    and not score_->below(ctx->peer_id, config_.score.zero);
+               },
+               config_.D - mesh_peers_.size())) {
+        addToMesh(ctx);
+      }
     }
   }
 
-  void TopicSubscriptions::onSelfSubscribed(bool self_subscribed) {
-    self_subscribed_ = self_subscribed;
-    if (!self_subscribed_) {
-      // remove the mesh
-      log_.debug("removing mesh for {}", topic_);
-      mesh_peers_.selectAll(
-          [this](const PeerContextPtr &p) { removeFromMesh(p); });
-      mesh_peers_.clear();
+  void TopicSubscriptions::unsubscribe() {
+    if (not self_subscribed_) {
+      return;
+    }
+    self_subscribed_ = false;
+    for (auto &ctx : mesh_peers_) {
+      sendPrune(ctx, true);
     }
+    mesh_peers_.clear();
   }
 
-  void TopicSubscriptions::onPeerSubscribed(const PeerContextPtr &p) {
-    assert(p->subscribed_to.count(topic_) != 0);
+  void TopicSubscriptions::onPeerSubscribed(const PeerContextPtr &ctx) {
+    assert(ctx->subscribed_to.count(topic_) != 0);
 
-    subscribed_peers_.insert(p);
+    subscribed_peers_.insert(ctx);
 
-    // announce the peer about messages available for the topic
-    for (const auto &[_, msg_id] : seen_cache_) {
-      p->message_builder->addIHave(topic_, msg_id);
+    if (ctx->isGossipsub() and mesh_peers_.size() < config_.D_min
+        and not mesh_peers_.contains(ctx)
+        and not explicit_peers_->contains(ctx->peer_id)
+        and not isBackoffWithSlack(ctx->peer_id)
+        and not score_->below(ctx->peer_id, config_.score.zero)) {
+      addToMesh(ctx);
     }
-    // will be sent on next heartbeat
-    connectivity_.peerIsWritable(p, false);
   }
 
-  void TopicSubscriptions::onPeerUnsubscribed(const PeerContextPtr &p) {
-    auto res = subscribed_peers_.erase(p->peer_id);
-    if (!res) {
-      res = mesh_peers_.erase(p->peer_id);
+  void TopicSubscriptions::onPeerUnsubscribed(const PeerContextPtr &ctx) {
+    subscribed_peers_.erase(ctx->peer_id);
+    if (fanout_) {
+      fanout_->peers.erase(ctx->peer_id);
+      if (fanout_->peers.empty()) {
+        fanout_.reset();
+      }
+    }
+    if (mesh_peers_.erase(ctx->peer_id)) {
+      score_->prune(ctx->peer_id, topic_);
+      updateBackoff(ctx->peer_id, config_.prune_backoff);
     }
-    dont_bother_until_.erase(p);
   }
 
-  void TopicSubscriptions::onGraft(const PeerContextPtr &p) {
-    auto res = mesh_peers_.find(p->peer_id);
+  void TopicSubscriptions::onGraft(const PeerContextPtr &ctx) {
+    auto res = mesh_peers_.find(ctx->peer_id);
     if (res) {
       // already there
       return;
     }
 
-    if (!subscribed_peers_.contains(p->peer_id)) {
-      // subscribe first
-      p->subscribed_to.insert(topic_);
-      onPeerSubscribed(p);
-    }
+    // implicit subscribe on graft
+    subscribed_peers_.insert(ctx);
 
-    bool mesh_is_full = (mesh_peers_.size() >= config_.D_max);
-
-    if (self_subscribed_ && !mesh_is_full) {
-      mesh_peers_.insert(p);
-      subscribed_peers_.erase(p->peer_id);
-    } else {
-      // we don't have mesh for the topic
-      p->message_builder->addPrune(topic_);
-      connectivity_.peerIsWritable(p, true);
+    bool prune = [&] {
+      if (self_subscribed_) {
+        return true;
+      }
+      if (explicit_peers_->contains(ctx->peer_id)) {
+        return true;
+      }
+      if (isBackoff(ctx->peer_id, std::chrono::milliseconds{0})) {
+        score_->addPenalty(ctx->peer_id, 1);
+        if (isBackoff(ctx->peer_id,
+                      config_.graft_flood_threshold - config_.prune_backoff)) {
+          score_->addPenalty(ctx->peer_id, 1);
+        }
+      }
+      if (score_->below(ctx->peer_id, config_.score.zero)) {
+        return true;
+      }
+      if (mesh_peers_.size() >= config_.D_max
+          and not outbound_peers_->contains(ctx->peer_id)) {
+        return true;
+      }
+      score_->graft(ctx->peer_id, topic_);
+      mesh_peers_.insert(ctx);
+      return false;
+    }();
+    if (prune) {
+      sendPrune(ctx, false);
     }
   }
 
-  void TopicSubscriptions::onPrune(const PeerContextPtr &p,
-                                   Time dont_bother_until) {
-    mesh_peers_.erase(p->peer_id);
-    if (p->subscribed_to.count(topic_) != 0) {
-      subscribed_peers_.insert(p);
-      dont_bother_until_.insert({p, dont_bother_until});
+  void TopicSubscriptions::onPrune(
+      const PeerContextPtr &ctx, std::optional<std::chrono::seconds> backoff) {
+    if (mesh_peers_.erase(ctx->peer_id)) {
+      score_->prune(ctx->peer_id, topic_);
     }
+    updateBackoff(ctx->peer_id, backoff.value_or(config_.prune_backoff));
   }
 
-  void TopicSubscriptions::addToMesh(const PeerContextPtr &p) {
-    assert(p->message_builder);
+  void TopicSubscriptions::addToMesh(const PeerContextPtr &ctx) {
+    assert(ctx->message_builder);
 
-    p->message_builder->addGraft(topic_);
-    connectivity_.peerIsWritable(p, false);
-    mesh_peers_.insert(p);
+    ctx->message_builder->addGraft(topic_);
+    connectivity_.peerIsWritable(ctx);
+    score_->graft(ctx->peer_id, topic_);
+    mesh_peers_.insert(ctx);
     log_.debug("peer {} added to mesh (size={}) for topic {}",
-               p->str,
+               ctx->str,
                mesh_peers_.size(),
                topic_);
   }
 
-  void TopicSubscriptions::removeFromMesh(const PeerContextPtr &p) {
-    assert(p->message_builder);
-
-    p->message_builder->addPrune(topic_);
-    connectivity_.peerIsWritable(p, false);
-    subscribed_peers_.insert(p);
+  void TopicSubscriptions::sendPrune(const PeerContextPtr &ctx,
+                                     bool unsubscribe) {
+    assert(ctx->message_builder);
+
+    auto backoff = unsubscribe ? config_.prune_backoff : config_.prune_backoff;
+    updateBackoff(ctx->peer_id, backoff);
+    ctx->message_builder->addPrune(
+        topic_,
+        ctx->isGossipsubv1_1() ? std::make_optional(backoff) : std::nullopt);
+    score_->prune(ctx->peer_id, topic_);
+    connectivity_.peerIsWritable(ctx);
     log_.debug("peer {} removed from mesh (size={}) for topic {}",
-               p->str,
+               ctx->str,
                mesh_peers_.size(),
                topic_);
   }
 
+  bool TopicSubscriptions::isBackoff(const PeerId &peer_id,
+                                     std::chrono::milliseconds slack) const {
+    auto it = dont_bother_until_.find(peer_id);
+    if (it == dont_bother_until_.end()) {
+      return false;
+    }
+    return it->second + slack > scheduler_->now();
+  }
+
+  bool TopicSubscriptions::isBackoffWithSlack(const PeerId &peer_id) const {
+    return isBackoff(peer_id,
+                     config_.backoff_slack * config_.heartbeat_interval_msec);
+  }
+
+  void TopicSubscriptions::updateBackoff(const PeerId &peer_id,
+                                         std::chrono::milliseconds duration) {
+    auto until = scheduler_->now() + duration;
+    auto [it, inserted] = dont_bother_until_.emplace(peer_id, until);
+    if (not inserted and it->second < until) {
+      it->second = until;
+    }
+  }
+
+  void TopicSubscriptions::emitGossip() {
+    auto &mesh = fanout_.has_value() ? fanout_->peers : mesh_peers_;
+    auto peers = choose_peers_->choose(
+        subscribed_peers_,
+        [&](const PeerContextPtr &ctx) {
+          return not mesh.contains(ctx)
+             and not explicit_peers_->contains(ctx->peer_id)
+             and not score_->below(ctx->peer_id,
+                                   config_.score.gossip_threshold);
+        },
+        [&](size_t n) {
+          return std::max(config_.D_lazy,
+                          static_cast<size_t>(static_cast<double>(n)
+                                              * config_.gossip_factor));
+        });
+    size_t message_count = 0;
+    for (auto &x : history_gossip_) {
+      message_count += x.size();
+    }
+    if (message_count == 0) {
+      return;
+    }
+    std::vector<MessageId> messages;
+    messages.reserve(message_count);
+    for (auto &x : history_gossip_) {
+      messages.insert(messages.end(), x.begin(), x.end());
+    }
+    std::ranges::shuffle(messages, gossip_random_);
+    for (auto &ctx : peers) {
+      std::span messages_span{messages};
+      if (messages.size() > config_.max_ihave_length) {
+        std::ranges::shuffle(messages, gossip_random_);
+        messages_span = messages_span.first(config_.max_ihave_length);
+      }
+      for (auto &message_id : messages_span) {
+        ctx->message_builder->addIHave(topic_, message_id);
+      }
+      connectivity_.peerIsWritable(ctx);
+    }
+  }
 }  // namespace libp2p::protocol::gossip
diff --git a/src/protocol/gossip/impl/topic_subscriptions.hpp b/src/protocol/gossip/impl/topic_subscriptions.hpp
index 83b4b2ad1..44d4be6b7 100644
--- a/src/protocol/gossip/impl/topic_subscriptions.hpp
+++ b/src/protocol/gossip/impl/topic_subscriptions.hpp
@@ -7,14 +7,21 @@
 #pragma once
 
 #include <deque>
+#include <random>
 
 #include <libp2p/log/sublogger.hpp>
 
 #include "peer_set.hpp"
 
-namespace libp2p::protocol::gossip {
+namespace libp2p::protocol::gossip::score {
+  class Score;
+}  // namespace libp2p::protocol::gossip::score
 
+namespace libp2p::protocol::gossip {
+  class ChoosePeers;
   class Connectivity;
+  class ExplicitPeers;
+  using score::Score;
 
   /// Per-topic subscriptions
   class TopicSubscriptions {
@@ -24,11 +31,18 @@ namespace libp2p::protocol::gossip {
     TopicSubscriptions(TopicId topic,
                        const Config &config,
                        Connectivity &connectivity,
+                       std::shared_ptr<basic::Scheduler> scheduler,
+                       std::shared_ptr<ChoosePeers> choose_peers,
+                       std::shared_ptr<ExplicitPeers> explicit_peers,
+                       std::shared_ptr<Score> score,
+                       std::shared_ptr<OutboundPeers> outbound_peers,
                        log::SubLogger &log);
 
-    /// Returns true if no peers subscribed and not self-subscribed and
+    /// Returns true if not self-subscribed and
     /// no fanout period at the moment (empty item may be erased)
-    bool empty() const;
+    bool isUsed() const;
+
+    bool isSubscribed() const;
 
     /// Forwards message to mesh members and announce to other subscribers
     void onNewMessage(const boost::optional<PeerContextPtr> &from,
@@ -39,8 +53,8 @@ namespace libp2p::protocol::gossip {
     /// Periodic job needed to update meshes and shift "I have" caches
     void onHeartbeat(Time now);
 
-    /// Local host subscribes or unsubscribes, this affects mesh
-    void onSelfSubscribed(bool self_subscribed);
+    void subscribe();
+    void unsubscribe();
 
     /// Remote peer subscribes to topic
     void onPeerSubscribed(const PeerContextPtr &p);
@@ -52,24 +66,42 @@ namespace libp2p::protocol::gossip {
     void onGraft(const PeerContextPtr &p);
 
     /// Remote peer kicks this host out of its mesh
-    void onPrune(const PeerContextPtr &p, Time dont_bother_until);
+    void onPrune(const PeerContextPtr &p,
+                 std::optional<std::chrono::seconds> backoff);
 
    private:
+    struct Fanout {
+      Time until;
+      PeerSet peers;
+    };
+
     /// Adds a peer to mesh
     void addToMesh(const PeerContextPtr &p);
 
     /// Removes a peer from mesh
-    void removeFromMesh(const PeerContextPtr &p);
+    void sendPrune(const PeerContextPtr &ctx, bool unsubscribe);
+
+    void emitGossip();
+    bool isBackoff(const PeerId &peer_id,
+                   std::chrono::milliseconds slack) const;
+    bool isBackoffWithSlack(const PeerId &peer_id) const;
+    void updateBackoff(const PeerId &peer_id,
+                       std::chrono::milliseconds duration);
 
     const TopicId topic_;
     const Config &config_;
     Connectivity &connectivity_;
+    std::shared_ptr<basic::Scheduler> scheduler_;
+    std::shared_ptr<ChoosePeers> choose_peers_;
+    std::shared_ptr<ExplicitPeers> explicit_peers_;
+    std::shared_ptr<Score> score_;
+    std::shared_ptr<OutboundPeers> outbound_peers_;
 
     /// This host subscribed to this topic or not, this affects mesh behavior
     bool self_subscribed_;
 
     /// Fanout period allows for publishing from this host without subscribing
-    Time fanout_period_ends_;
+    std::optional<Fanout> fanout_;
 
     /// Peers subscribed to this topic, but not mesh members
     PeerSet subscribed_peers_;
@@ -77,13 +109,17 @@ namespace libp2p::protocol::gossip {
     /// Mesh members to whom messages are forwarded in push manner
     PeerSet mesh_peers_;
 
-    /// "I have" notifications for new subscribers aka seen messages cache
-    std::deque<std::pair<Time, MessageId>> seen_cache_;
+    /// "I have" notifications for new subscribers
+    std::deque<std::deque<MessageId>> history_gossip_;
 
     /// Prune backoff times per peer
-    std::unordered_map<PeerContextPtr, Time> dont_bother_until_;
+    std::unordered_map<PeerId, Time> dont_bother_until_;
 
     log::SubLogger &log_;
+
+    std::default_random_engine gossip_random_;
+
+    size_t heartbeat_ticks_ = 0;
   };
 
 }  // namespace libp2p::protocol::gossip
diff --git a/src/protocol/gossip/protobuf/rpc.proto b/src/protocol/gossip/protobuf/rpc.proto
index 6de40af24..2bfa62783 100644
--- a/src/protocol/gossip/protobuf/rpc.proto
+++ b/src/protocol/gossip/protobuf/rpc.proto
@@ -28,6 +28,7 @@ message ControlMessage {
 	repeated ControlIWant iwant = 2;
 	repeated ControlGraft graft = 3;
 	repeated ControlPrune prune = 4;
+	repeated ControlIDontWant idontwant = 5;
 }
 
 message ControlIHave {
@@ -53,3 +54,7 @@ message ControlPrune {
 	repeated PeerInfo peers = 2;
 	optional uint64 backoff = 3;
 }
+
+message ControlIDontWant {
+	repeated bytes message_ids = 1;
+}
diff --git a/test/libp2p/protocol/gossip/CMakeLists.txt b/test/libp2p/protocol/gossip/CMakeLists.txt
index dcd6aba15..19b6baa09 100644
--- a/test/libp2p/protocol/gossip/CMakeLists.txt
+++ b/test/libp2p/protocol/gossip/CMakeLists.txt
@@ -20,3 +20,15 @@ target_link_libraries(gossip_local_subs_test
     p2p_gossip
     p2p_testutil_peer
     )
+
+addtest(gossip_mock_test
+    gossip_mock_test.cpp
+    )
+target_link_libraries(gossip_mock_test
+    p2p_basic_scheduler
+    p2p_gossip
+    p2p_manual_scheduler_backend
+    p2p_message_read_writer
+    p2p_stream_pair
+    p2p_testutil_peer
+    )
diff --git a/test/libp2p/protocol/gossip/gossip_mock_test.cpp b/test/libp2p/protocol/gossip/gossip_mock_test.cpp
new file mode 100644
index 000000000..d6255e5c5
--- /dev/null
+++ b/test/libp2p/protocol/gossip/gossip_mock_test.cpp
@@ -0,0 +1,517 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <generated/protocol/gossip/protobuf/rpc.pb.h>
+#include <gtest/gtest.h>
+#include <libp2p/basic/message_read_writer_uvarint.hpp>
+#include <libp2p/basic/scheduler/manual_scheduler_backend.hpp>
+#include <libp2p/basic/scheduler/scheduler_impl.hpp>
+#include <libp2p/connection/stream_pair.hpp>
+#include <libp2p/protocol/gossip/gossip.hpp>
+#include <qtils/bytestr.hpp>
+
+#include "mock/libp2p/crypto/crypto_provider.hpp"
+#include "mock/libp2p/crypto/key_marshaller_mock.hpp"
+#include "mock/libp2p/host/host_mock.hpp"
+#include "mock/libp2p/peer/address_repository_mock.hpp"
+#include "mock/libp2p/peer/identity_manager_mock.hpp"
+#include "mock/libp2p/peer/peer_repository_mock.hpp"
+#include "testutil/libp2p/peer.hpp"
+#include "testutil/prepare_loggers.hpp"
+
+using libp2p::Bytes;
+using libp2p::BytesIn;
+using libp2p::HostMock;
+using libp2p::PeerId;
+using libp2p::PeerInfo;
+using libp2p::ProtocolName;
+using libp2p::ProtocolPredicate;
+using libp2p::StreamAndProtocol;
+using libp2p::StreamAndProtocolCb;
+using libp2p::StreamAndProtocolOrErrorCb;
+using libp2p::StreamProtocols;
+using libp2p::basic::ManualSchedulerBackend;
+using libp2p::basic::MessageReadWriterUvarint;
+using libp2p::basic::SchedulerImpl;
+using libp2p::connection::Stream;
+using libp2p::crypto::CryptoProviderMock;
+using libp2p::crypto::marshaller::KeyMarshallerMock;
+using libp2p::peer::AddressRepositoryMock;
+using libp2p::peer::IdentityManagerMock;
+using libp2p::peer::PeerRepositoryMock;
+using libp2p::protocol::gossip::Gossip;
+using libp2p::protocol::gossip::PeerKind;
+using libp2p::protocol::gossip::TopicId;
+using testing::_;
+using testing::Return;
+using testing::ReturnRef;
+using testutil::randomPeerId;
+
+using MessageIds = std::vector<uint8_t>;
+
+inline Bytes encodeMessageId(uint8_t i) {
+  return Bytes{i};
+}
+inline uint8_t decodeMessageId(std::string_view data) {
+  return data.at(0);
+}
+
+struct Received {
+  std::vector<bool> subscriptions{};
+  MessageIds messages{};
+  std::vector<bool> graft{};
+  MessageIds ihave{};
+  MessageIds iwant{};
+  MessageIds idontwant{};
+};
+
+struct MockPeer {
+  MockPeer(PeerId peer_id, ProtocolName version, std::shared_ptr<Stream> writer)
+      : peer_id_{std::move(peer_id)},
+        version_{std::move(version)},
+        writer_{std::move(writer)},
+        framing_{std::make_shared<MessageReadWriterUvarint>(writer_)} {}
+
+  void expect(Received expected) {
+    EXPECT_EQ(received_.subscriptions, expected.subscriptions);
+    EXPECT_EQ(received_.messages, expected.messages);
+    EXPECT_EQ(received_.graft, expected.graft);
+    EXPECT_EQ(received_.ihave, expected.ihave);
+    EXPECT_EQ(received_.iwant, expected.iwant);
+    EXPECT_EQ(received_.idontwant, expected.idontwant);
+    received_ = {};
+  }
+
+  void write(const auto &f) {
+    pubsub::pb::RPC rpc;
+    f(rpc);
+    Bytes buffer;
+    buffer.resize(rpc.ByteSizeLong());
+    rpc.SerializeToArray(buffer.data(), buffer.size());
+    framing_->write(
+        buffer, [](outcome::result<size_t> r) { EXPECT_TRUE(r.has_value()); });
+  }
+
+  PeerId peer_id_;
+  ProtocolName version_;
+  std::shared_ptr<Stream> writer_;
+  std::shared_ptr<MessageReadWriterUvarint> framing_;
+  Received received_;
+};
+
+struct GossipMockTest : testing::Test {
+  PeerId gossip_peer_id_{randomPeerId()};
+  TopicId topic1{"topic1"};
+  std::unordered_map<PeerId, std::shared_ptr<MockPeer>> peers_;
+  libp2p::protocol::gossip::Config config_;
+  std::optional<StreamAndProtocolCb> host_handler_;
+  std::shared_ptr<ManualSchedulerBackend> scheduler_backend_ =
+      std::make_shared<ManualSchedulerBackend>();
+  std::shared_ptr<SchedulerImpl> scheduler_ = std::make_shared<SchedulerImpl>(
+      scheduler_backend_, SchedulerImpl::Config{});
+  libp2p::event::Bus bus_;
+  std::shared_ptr<PeerRepositoryMock> peer_repo_ =
+      std::make_shared<PeerRepositoryMock>();
+  std::shared_ptr<AddressRepositoryMock> address_repo_ =
+      std::make_shared<AddressRepositoryMock>();
+  std::shared_ptr<HostMock> host_ = std::make_shared<HostMock>();
+  std::shared_ptr<IdentityManagerMock> idmgr_ =
+      std::make_shared<IdentityManagerMock>();
+  std::shared_ptr<CryptoProviderMock> crypto_provider_ =
+      std::make_shared<CryptoProviderMock>();
+  std::shared_ptr<KeyMarshallerMock> key_marshaller_ =
+      std::make_shared<KeyMarshallerMock>();
+  std::shared_ptr<Gossip> gossip_;
+
+  static void SetUpTestCase() {
+    testutil::prepareLoggers();
+  }
+
+  void SetUp() override {
+    config_.D_min = 1;
+    config_.D = 1;
+    config_.D_lazy = 1;
+    config_.flood_publish = false;
+  }
+
+  void setup() {
+    EXPECT_CALL(*host_, getBus()).WillRepeatedly(ReturnRef(bus_));
+    EXPECT_CALL(*host_, getPeerRepository())
+        .WillRepeatedly(ReturnRef(*peer_repo_));
+    EXPECT_CALL(*peer_repo_, getAddressRepository())
+        .WillRepeatedly(ReturnRef(*address_repo_));
+    EXPECT_CALL(*address_repo_, updateAddresses(_, _))
+        .WillRepeatedly(Return(outcome::success()));
+    EXPECT_CALL(*host_, getPeerInfo())
+        .WillRepeatedly(Return(PeerInfo{gossip_peer_id_, {}}));
+    EXPECT_CALL(*host_, setProtocolHandler(_, _, _))
+        .WillRepeatedly([&](StreamProtocols,
+                            StreamAndProtocolCb cb,
+                            ProtocolPredicate) { host_handler_ = cb; });
+    EXPECT_CALL(*host_, newStream(_, _, _))
+        .WillRepeatedly([&](const PeerInfo &info,
+                            StreamProtocols,
+                            StreamAndProtocolOrErrorCb cb) {
+          auto &peer_id = info.id;
+          auto peer = peers_.at(peer_id);
+          auto [stream1, stream2] = libp2p::connection::streamPair(
+              scheduler_, peer_id, gossip_peer_id_);
+          read(peer, std::make_shared<MessageReadWriterUvarint>(stream2));
+          scheduler_->schedule([cb, stream1, peer] {
+            cb(StreamAndProtocol{stream1, peer->version_});
+          });
+        });
+    gossip_ = libp2p::protocol::gossip::create(
+        scheduler_, host_, idmgr_, crypto_provider_, key_marshaller_, config_);
+    gossip_->setMessageIdFn(
+        [](BytesIn from, BytesIn seqno, Bytes data) { return data; });
+    gossip_->start();
+  }
+
+  void TearDown() override {
+    for (auto &p : peers_) {
+      p.second->expect({});
+    }
+  }
+
+  auto connect(PeerKind peer_kind) {
+    auto version_it =
+        std::ranges::find_if(config_.protocol_versions,
+                             [&](auto &p) { return p.second == peer_kind; });
+    if (version_it == config_.protocol_versions.end()) {
+      throw std::logic_error{"PeerKind"};
+    }
+    auto &version = version_it->first;
+    auto peer_id = randomPeerId();
+    auto [stream1, stream2] =
+        libp2p::connection::streamPair(scheduler_, peer_id, gossip_peer_id_);
+    auto peer = std::make_shared<MockPeer>(peer_id, version, stream1);
+    peers_.emplace(peer_id, peer);
+    scheduler_->schedule([this, stream2, version] {
+      host_handler_.value()({stream2, version});
+    });
+    scheduler_backend_->callDeferred();
+    return peer;
+  }
+
+  void read(std::shared_ptr<MockPeer> peer,
+            std::shared_ptr<MessageReadWriterUvarint> framing) {
+    framing->read([this, peer, framing](
+                      outcome::result<std::shared_ptr<Bytes>> frame_res) {
+      if (frame_res.has_error()) {
+        return;
+      }
+      auto &frame = frame_res.value();
+      pubsub::pb::RPC rpc;
+      EXPECT_TRUE(rpc.ParseFromArray(frame->data(), frame->size()));
+      for (auto &sub : rpc.subscriptions()) {
+        EXPECT_EQ(sub.topicid(), topic1);
+        peer->received_.subscriptions.emplace_back(sub.subscribe());
+      }
+      for (auto &pub : rpc.publish()) {
+        peer->received_.messages.emplace_back(decodeMessageId(pub.data()));
+      }
+      for (auto &graft : rpc.control().graft()) {
+        EXPECT_EQ(graft.topicid(), topic1);
+        peer->received_.graft.emplace_back(true);
+      }
+      EXPECT_EQ(rpc.control().prune().size(), 0);
+      for (auto &ihave : rpc.control().ihave()) {
+        EXPECT_EQ(ihave.topicid(), topic1);
+        for (auto &id : ihave.messageids()) {
+          peer->received_.ihave.emplace_back(decodeMessageId(id));
+        }
+      }
+      for (auto &iwant : rpc.control().iwant()) {
+        for (auto &id : iwant.messageids()) {
+          peer->received_.iwant.emplace_back(decodeMessageId(id));
+        }
+      }
+      for (auto &idontwant : rpc.control().idontwant()) {
+        for (auto &id : idontwant.message_ids()) {
+          peer->received_.idontwant.emplace_back(decodeMessageId(id));
+        }
+      }
+      read(peer, framing);
+    });
+  }
+
+  auto subscribe() {
+    auto sub = gossip_->subscribe({topic1}, [](Gossip::SubscriptionData) {});
+    scheduler_backend_->callDeferred();
+    for (auto &p : peers_) {
+      p.second->expect({.subscriptions = {true}});
+    }
+    return sub;
+  }
+
+  void subscribe(MockPeer &peer, bool subscribe) {
+    peer.write([&](pubsub::pb::RPC &rpc) {
+      auto *sub = rpc.add_subscriptions();
+      sub->set_topicid(topic1);
+      sub->set_subscribe(subscribe);
+    });
+    scheduler_backend_->callDeferred();
+  }
+
+  void publish(uint8_t i) {
+    gossip_->publish(topic1, {i});
+    scheduler_backend_->callDeferred();
+  }
+
+  void publish(MockPeer &peer,
+               uint8_t i,
+               std::optional<PeerId> author = std::nullopt) {
+    peer.write([&](pubsub::pb::RPC &rpc) {
+      auto *pub = rpc.add_publish();
+      *pub->mutable_from() =
+          qtils::byte2str(author.value_or(peer.peer_id_).toVector());
+      pub->set_seqno("");
+      pub->set_topic(topic1);
+      *pub->mutable_data() = qtils::byte2str(encodeMessageId(i));
+    });
+    scheduler_backend_->callDeferred();
+  }
+
+  void ihave(MockPeer &peer, uint8_t i) {
+    peer.write([&](pubsub::pb::RPC &rpc) {
+      auto *ihave = rpc.mutable_control()->add_ihave();
+      ihave->set_topicid(topic1);
+      *ihave->add_messageids() = qtils::byte2str(encodeMessageId(i));
+    });
+    scheduler_backend_->callDeferred();
+  }
+
+  void iwant(MockPeer &peer, uint8_t i) {
+    peer.write([&](pubsub::pb::RPC &rpc) {
+      auto *iwant = rpc.mutable_control()->add_iwant();
+      *iwant->add_messageids() = qtils::byte2str(encodeMessageId(i));
+    });
+    scheduler_backend_->callDeferred();
+  }
+
+  void idontwant(MockPeer &peer, uint8_t i) {
+    peer.write([&](pubsub::pb::RPC &rpc) {
+      auto *idontwant = rpc.mutable_control()->add_idontwant();
+      *idontwant->add_message_ids() = qtils::byte2str(encodeMessageId(i));
+    });
+    scheduler_backend_->callDeferred();
+  }
+
+  void heartbeat() {
+    scheduler_backend_->shift(config_.heartbeat_interval_msec);
+  }
+};
+
+/**
+ * notify peers when subscribing and unsubscribing.
+ */
+TEST_F(GossipMockTest, SubscribeUnsubscribe) {
+  setup();
+  auto peer1 = connect(PeerKind::Floodsub);
+
+  auto sub = subscribe();
+
+  sub.cancel();
+  scheduler_backend_->callDeferred();
+  peer1->expect({.subscriptions = {false}});
+}
+
+/**
+ * publish to subscribed peers only.
+ * don't publish until peers subscribe.
+ * don't publish after peers unsubscribe.
+ */
+TEST_F(GossipMockTest, PublishToFloodsub) {
+  setup();
+  auto peer1 = connect(PeerKind::Floodsub);
+
+  publish(1);
+  peer1->expect({});
+
+  subscribe(*peer1, true);
+  publish(2);
+  peer1->expect({.messages = {2}});
+
+  subscribe(*peer1, false);
+  publish(3);
+  peer1->expect({});
+}
+
+/**
+ * forwards message to floodsub peers except:
+ *  - peer who is not subscribed
+ *  - peer who sent the message
+ *  - message author
+ * don't forward same message more than once.
+ */
+TEST_F(GossipMockTest, ForwardToFloodsub) {
+  setup();
+  auto peer1 = connect(PeerKind::Floodsub);
+  auto peer2 = connect(PeerKind::Floodsub);
+  auto peer3 = connect(PeerKind::Floodsub);
+  auto peer4 = connect(PeerKind::Floodsub);
+
+  auto sub = subscribe();
+  subscribe(*peer1, true);
+  subscribe(*peer2, true);
+  subscribe(*peer3, true);
+  publish(*peer2, 1, peer1->peer_id_);
+  peer3->expect({.messages = {1}});
+
+  publish(*peer2, 1, peer1->peer_id_);
+}
+
+/**
+ * publish to fanout peers.
+ * must publish to same initially chosen fanout peers.
+ * don't forward messages to fanout peers.
+ */
+TEST_F(GossipMockTest, PublishToFanout) {
+  setup();
+  auto peer1 = connect(PeerKind::Gossipsub);
+  auto peer2 = connect(PeerKind::Gossipsub);
+  uint8_t i = 1;
+
+  subscribe(*peer1, true);
+  subscribe(*peer2, true);
+  publish(i);
+  if (peer1->received_.messages.empty()) {
+    std::swap(peer1, peer2);
+  }
+  peer1->expect({.messages = {i}});
+  i++;
+  while (i < 30) {
+    publish(i);
+    peer1->expect({.messages = {i}});
+    i++;
+  }
+
+  publish(*peer2, i);
+}
+
+/**
+ * notify peers when grafting.
+ * publish to grafted peers.
+ * forward to grafted peers.
+ */
+TEST_F(GossipMockTest, PublishToMesh) {
+  setup();
+  auto peer1 = connect(PeerKind::Gossipsub);
+  auto peer2 = connect(PeerKind::Gossipsub);
+
+  auto sub = subscribe();
+  subscribe(*peer1, true);
+  subscribe(*peer2, true);
+  peer1->expect({.graft = {true}});
+
+  publish(1);
+  peer1->expect({.messages = {1}});
+
+  publish(*peer2, 2, randomPeerId());
+  peer1->expect({.messages = {2}});
+}
+
+/**
+ * publish to all peers.
+ * don't forward to all peers.
+ */
+TEST_F(GossipMockTest, FloodPublish) {
+  config_.flood_publish = true;
+  setup();
+  auto peer1 = connect(PeerKind::Gossipsub);
+  auto peer2 = connect(PeerKind::Gossipsub);
+
+  auto sub = subscribe();
+  subscribe(*peer1, true);
+  subscribe(*peer2, true);
+  publish(1);
+  publish(*peer2, 2, randomPeerId());
+  peer1->expect({.messages = {1, 2}, .graft = {true}});
+  peer2->expect({.messages = {1}});
+}
+
+/**
+ * gossip recent messages to random peers.
+ * don't gossip to mesh or fanout peers.
+ */
+TEST_F(GossipMockTest, Gossip) {
+  setup();
+  auto peer1 = connect(PeerKind::Gossipsub);
+  auto peer2 = connect(PeerKind::Gossipsub);
+  auto peer3 = connect(PeerKind::Gossipsub);
+
+  auto sub = subscribe();
+  subscribe(*peer1, true);
+  subscribe(*peer2, true);
+  subscribe(*peer3, true);
+  publish(1);
+  peer1->expect({.messages = {1}, .graft = {true}});
+
+  heartbeat();
+  if (peer2->received_.ihave.empty()) {
+    std::swap(peer2, peer3);
+  }
+  peer2->expect({.ihave = {1}});
+}
+
+/**
+ * send iwant after receiving ihave.
+ * don't send iwant until subscribed to topic.
+ * don't send iwant after receiving message.
+ */
+TEST_F(GossipMockTest, IhaveIwant) {
+  setup();
+  auto peer1 = connect(PeerKind::Gossipsub);
+
+  ihave(*peer1, 1);
+  peer1->expect({});
+
+  auto sub = subscribe();
+  ihave(*peer1, 1);
+  peer1->expect({.iwant = {1}});
+
+  publish(*peer1, 1);
+  ihave(*peer1, 1);
+
+  iwant(*peer1, 1);
+  peer1->expect({.messages = {1}});
+}
+
+/**
+ * send idontwant after receiving message.
+ * send idontwant to mesh peers.
+ * send idontwant to cancel pending iwant requests.
+ * don't reply to iwant from peer after receiving idontwant.
+ * don't forward message to peer after receiving idontwant.
+ */
+TEST_F(GossipMockTest, Idontwant) {
+  config_.idontwant_message_size_threshold = 1;
+  setup();
+  auto peer1 = connect(PeerKind::Gossipsubv1_2);
+  auto peer2 = connect(PeerKind::Gossipsubv1_2);
+
+  auto sub = subscribe();
+  subscribe(*peer1, true);
+  subscribe(*peer2, true);
+  publish(*peer2, 1);
+  peer1->expect({.messages = {1}, .graft = {1}, .idontwant = {1}});
+
+  ihave(*peer2, 2);
+  peer2->expect({.iwant = {2}});
+  publish(*peer1, 2);
+  peer1->expect({.idontwant = {2}});
+  peer2->expect({.idontwant = {2}});
+
+  idontwant(*peer2, 2);
+  iwant(*peer2, 2);
+  peer2->expect({});
+
+  idontwant(*peer2, 3);
+  publish(*peer1, 3);
+  peer1->expect({.idontwant = {3}});
+  peer2->expect({});
+}
diff --git a/test/mock/libp2p/crypto/crypto_provider.hpp b/test/mock/libp2p/crypto/crypto_provider.hpp
new file mode 100644
index 000000000..b04624bf2
--- /dev/null
+++ b/test/mock/libp2p/crypto/crypto_provider.hpp
@@ -0,0 +1,45 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <libp2p/crypto/crypto_provider.hpp>
+
+#include <gmock/gmock.h>
+
+namespace libp2p::crypto {
+  class CryptoProviderMock : public CryptoProvider {
+   public:
+    MOCK_METHOD(outcome::result<KeyPair>,
+                generateKeys,
+                (Key::Type, common::RSAKeyType),
+                (const, override));
+
+    MOCK_METHOD(outcome::result<PublicKey>,
+                derivePublicKey,
+                (const PrivateKey &),
+                (const, override));
+
+    MOCK_METHOD(outcome::result<Buffer>,
+                sign,
+                (BytesIn, const PrivateKey &),
+                (const, override));
+
+    MOCK_METHOD(outcome::result<bool>,
+                verify,
+                (BytesIn, BytesIn, const PublicKey &),
+                (const, override));
+    MOCK_METHOD(outcome::result<EphemeralKeyPair>,
+                generateEphemeralKeyPair,
+                (common::CurveType),
+                (const, override));
+
+    MOCK_METHOD((outcome::result<std::pair<StretchedKey, StretchedKey>>),
+                stretchKey,
+                (common::CipherType, common::HashType, const Buffer &),
+                (const, override));
+  };
+}  // namespace libp2p::crypto