Don't send old squeaks in subscribe response in new proto version (#1423)

* Don't send old squeaks in subscribe response in new proto version

* Remove old comment in update subscription method

requirements-itest.txt

@@ -3,4 +3,4 @@ grpcio==1.39.0
 grpcio-tools==1.39.0
 importlib_resources==1.4.0
 pytest==6.2.5
-squeakpy==0.7.5
+squeakpy==0.7.6

requirements.txt

@@ -16,5 +16,5 @@ python-bitcoinlib==0.11.0
 pyzmq==22.3.0
 requests==2.26.0
 SQLAlchemy==1.3.24
-squeakpy==0.7.5
+squeakpy==0.7.6
 typed-config==0.2.5

squeaknode/network/connection.py

@@ -34,7 +34,6 @@ from squeak.messages import MSG_SECRET_KEY
 from squeak.messages import msg_secretkey
 from squeak.messages import MSG_SQUEAK
 from squeak.messages import msg_squeak
-from squeak.messages import msg_subscribe
 from squeak.net import CInterested
 from squeak.net import CInv
 
@@ -111,10 +110,7 @@ class Connection(object):
 
     def update_subscription(self):
         locator = self.squeak_controller.get_interested_locator()
-        subscribe_msg = msg_subscribe(
-            locator=locator,
-        )
-        self.peer.send_msg(subscribe_msg)
+        self.peer.update_local_subscription(locator)
 
     def update_addrs(self):
         getaddr_msg = msg_getaddr()
@@ -244,7 +240,8 @@ class Connection(object):
         )
 
     def handle_subscribe(self, msg):
-        self._send_reply_invs(msg.locator)
+        if msg.protover < 60003:
+            self._send_reply_invs(msg.locator)
         self.peer.set_remote_subscription(msg)
 
     def _send_reply_invs(self, locator):

squeaknode/network/network_manager.py

@@ -27,6 +27,7 @@ from typing import Optional
 
 import squeak.params
 from squeak.messages import MsgSerializable
+from squeak.net import CSqueakLocator
 
 from squeaknode.core.peer_address import PeerAddress
 from squeaknode.network.connection_manager import ConnectionManager
@@ -117,6 +118,15 @@ class NetworkManager(object):
                     peer,
                 ))
 
+    def update_local_subscriptions(self, locator: CSqueakLocator) -> None:
+        for peer in self.connection_manager.peers:
+            try:
+                peer.update_local_subscription(locator)
+            except Exception:
+                logger.exception("Failed to update local subcription with peer: {}".format(
+                    peer,
+                ))
+
     @property
     def local_address(self) -> PeerAddress:
         return PeerAddress(

squeaknode/network/peer.py

@@ -28,9 +28,12 @@ from io import BytesIO
 from bitcoin.core.serialize import SerializationTruncationError
 from bitcoin.net import CAddress
 from squeak.core import CSqueak
+from squeak.messages import msg_getsqueaks
+from squeak.messages import msg_subscribe
 from squeak.messages import msg_verack
 from squeak.messages import msg_version
 from squeak.messages import MsgSerializable
+from squeak.net import CSqueakLocator
 
 from squeaknode.core.peer_address import PeerAddress
 from squeaknode.core.util import generate_version_nonce
@@ -278,6 +281,16 @@ class Peer(object):
                 return True
         return False
 
+    def update_local_subscription(self, locator: CSqueakLocator):
+        getsqueaks_msg = msg_getsqueaks(
+            locator=locator,
+        )
+        subscribe_msg = msg_subscribe(
+            locator=locator,
+        )
+        self.send_msg(getsqueaks_msg)
+        self.send_msg(subscribe_msg)
+
     def on_peer_updated(self):
         logger.debug('on_peer_updated: {}'.format(self))
         self.peer_changed_listener.handle_new_item(self)

squeaknode/node/squeak_controller.py

@@ -33,7 +33,6 @@ from squeak.core.signing import CSigningKey
 from squeak.core.signing import CSqueakAddress
 from squeak.messages import msg_getdata
 from squeak.messages import msg_getsqueaks
-from squeak.messages import msg_subscribe
 from squeak.messages import MsgSerializable
 from squeak.net import CInterested
 from squeak.net import CInv
@@ -687,7 +686,7 @@ class SqueakController:
             reply_to_hash,
         )
 
-    def get_interested_locator(self):
+    def get_interested_locator(self) -> CSqueakLocator:
         block_range = self.get_block_range()
         followed_addresses = self.get_followed_addresses()
         if len(followed_addresses) == 0:
@@ -830,10 +829,7 @@ class SqueakController:
 
     def update_subscriptions(self):
         locator = self.get_interested_locator()
-        subscribe_msg = msg_subscribe(
-            locator=locator,
-        )
-        self.broadcast_msg(subscribe_msg)
+        self.network_manager.update_local_subscriptions(locator)
 
     def subscribe_received_offers_for_squeak(self, squeak_hash: bytes, stopped: threading.Event):
         for received_offer in self.new_received_offer_listener.yield_items(stopped):