Merge pull request #4253 from ichorid/f_mdversionfix

Stop old entries from being pushed back by gossip

Tribler/Core/Modules/MetadataStore/OrmBindings/channel_node.py

@@ -93,6 +93,9 @@ def __init__(self, *args, **kwargs):
             if "id_" not in kwargs:
                 kwargs["id_"] = self._clock.tick()
 
+            if "timestamp" not in kwargs:
+                kwargs["timestamp"] = kwargs["id_"]
+
             if not private_key_override and not skip_key_check:
                 # No key/signature given, sign with our own key.
                 if ("signature" not in kwargs) and \

Tribler/Core/Modules/MetadataStore/store.py

@@ -216,8 +216,6 @@ def process_squashed_mdblob(self, chunk_data):
     # Can't use db_session wrapper here, performance drops 10 times! Pony bug!
     def process_payload(self, payload):
         with db_session:
-            if self.ChannelNode.exists(signature=payload.signature):
-                return None, GOT_SAME_VERSION
 
             if payload.metadata_type == DELETED:
                 # We only allow people to delete their own entries, thus PKs must match
@@ -228,44 +226,33 @@ def process_payload(self, payload):
                     return None, DELETED_METADATA
                 else:
                     return None, NO_ACTION
+
+            # Check if we already got an older version of the same node that we can update
+            # TODO: optimize this with a single UPSERT-style query
+            local_node = self.ChannelNode.get(public_key=payload.public_key, id_=payload.id_)
+            if local_node:
+                if local_node.timestamp < payload.timestamp:
+                    local_node.set(**payload.to_dict())
+                    return local_node, UPDATED_OUR_VERSION
+                elif local_node.timestamp > payload.timestamp:
+                    return local_node, GOT_NEWER_VERSION
+                return local_node, GOT_SAME_VERSION
+
+            # Get the corresponding channel from local database to see if we really need to update our
+            # local contents of the channel by comparing the channel's local_version with the payload's timestamp.
+            # This check is necessary to prevent other peers pushing deleted entries into the
+            # channels we are subscribed to.
+            # If local channel version is 0, we are still in preview mode and willing to collect everything.
+            channel = self.ChannelMetadata.get(public_key=payload.public_key, id_=payload.origin_id)
+            if channel and (channel.local_version != 0) and (payload.timestamp <= channel.local_version):
+                return None, NO_ACTION
             elif payload.metadata_type == REGULAR_TORRENT:
                 return self.TorrentMetadata.from_payload(payload), UNKNOWN_TORRENT
             elif payload.metadata_type == CHANNEL_TORRENT:
-                return self.update_channel_info(payload)
+                return self.ChannelMetadata.from_payload(payload), UNKNOWN_CHANNEL
 
             return None, NO_ACTION
 
-    @db_session
-    def update_channel_info(self, payload):
-        """
-        We received some channel metadata, possibly over the network.
-        Validate the signature, update the local metadata store and put in at the beginning of the download queue
-        if necessary.
-        :param payload: The channel metadata, in serialized form.
-        :returns (metadata, status): tuple consisting of possibly newer metadata and result status
-        """
-
-        channel = self.ChannelMetadata.get_channel_with_id(payload.public_key)
-        if channel:
-            if payload.timestamp > channel.timestamp:
-                # Update the channel that is already there.
-                self._logger.info("Updating channel metadata %s ts %s->%s", str(channel.public_key).encode("hex"),
-                                  str(channel.timestamp), str(payload.timestamp))
-                channel.set(**ChannelMetadataPayload.to_dict(payload))
-                status = UPDATED_OUR_VERSION
-            elif payload.timestamp == channel.timestamp:
-                status = GOT_SAME_VERSION
-            else:
-                status = GOT_NEWER_VERSION
-
-        else:
-            status = UNKNOWN_CHANNEL
-            # Add new channel object to DB
-            channel = self.ChannelMetadata.from_payload(payload)
-
-        #TODO: handle the case where the local version is the same as the new one and is not seeded
-        return channel, status
-
     @db_session
     def get_my_channel(self):
         return self.ChannelMetadata.get_channel_with_id(self.my_key.pub().key_to_bin()[10:])

Tribler/Core/Modules/restapi/downloads_endpoint.py

@@ -18,6 +18,7 @@
 import Tribler.Core.Utilities.json_util as json
 from Tribler.Core.DownloadConfig import DownloadStartupConfig
 from Tribler.Core.Modules.MetadataStore.serialization import ChannelMetadataPayload
+from Tribler.Core.Modules.MetadataStore.store import UNKNOWN_CHANNEL
 from Tribler.Core.Modules.restapi.util import return_handled_exception
 from Tribler.Core.Utilities.torrent_utils import get_info_from_handle
 from Tribler.Core.Utilities.utilities import unichar_string
@@ -349,8 +350,8 @@ def on_error(error):
                     return json.dumps({"error": "Metadata has invalid signature"})
 
                 with db_session:
-                    channel, _ = self.session.lm.mds.process_payload(payload)
-                    if channel and not channel.subscribed and channel.local_version < channel.timestamp:
+                    channel, status = self.session.lm.mds.process_payload(payload)
+                    if channel and not channel.subscribed and status == UNKNOWN_CHANNEL:
                         channel.subscribed = True
                         download, _ = self.session.lm.gigachannel_manager.download_channel(channel)
                     else:

Tribler/Test/Core/Modules/MetadataStore/test_channel_download.py

@@ -56,5 +56,5 @@ def test_channel_update_and_download(self):
             # There should be 4 torrents + 1 channel torrent
             channel2 = self.session.lm.mds.ChannelMetadata.get_channel_with_id(payload.public_key)
             self.assertEqual(5, len(list(self.session.lm.mds.TorrentMetadata.select())))
-            self.assertEqual(9, channel2.timestamp)
+            self.assertEqual(1551110113009, channel2.timestamp)
             self.assertEqual(channel2.timestamp, channel2.local_version)

Tribler/Test/Core/Modules/MetadataStore/test_channel_metadata.py

@@ -137,9 +137,9 @@ def test_process_channel_metadata_payload(self):
 
         # Check that we always take the latest version
         channel_metadata.timestamp -= 1
-        self.assertEqual(channel_metadata.timestamp, 6)
+        self.assertEqual(channel_metadata.timestamp, 1551110113006)
         channel_metadata = self.mds.ChannelMetadata.process_channel_metadata_payload(payload)
-        self.assertEqual(channel_metadata.timestamp, 7)
+        self.assertEqual(channel_metadata.timestamp, 1551110113007)
         self.assertEqual(len(self.mds.ChannelMetadata.select()), 1)
 
     @db_session

Tribler/Test/Core/Modules/MetadataStore/test_store.py

@@ -50,7 +50,6 @@ def tearDown(self):
         self.mds.shutdown()
         yield super(TestMetadataStore, self).tearDown()
 
-
     def test_store_clock(self):
         my_key = default_eccrypto.generate_key(u"curve25519")
         mds2 = MetadataStore(os.path.join(self.session_base_dir, 'test.db'), self.session_base_dir, my_key)
@@ -146,7 +145,7 @@ def test_process_channel_dir(self):
         self.assertFalse(channel.contents_list)
         self.mds.process_channel_dir(self.CHANNEL_DIR, channel.public_key)
         self.assertEqual(len(channel.contents_list), 3)
-        self.assertEqual(channel.timestamp, 7)
+        self.assertEqual(channel.timestamp, 1551110113007)
         self.assertEqual(channel.local_version, channel.timestamp)
 
     @db_session
@@ -159,7 +158,7 @@ def get_payloads(entity_class):
 
         _, node_payload, node_deleted_payload = get_payloads(self.mds.ChannelNode)
 
-        self.assertEqual((None, GOT_SAME_VERSION), self.mds.process_payload(node_payload))
+        self.assertEqual((self.mds.ChannelNode[1], GOT_SAME_VERSION), self.mds.process_payload(node_payload))
         self.assertEqual((None, DELETED_METADATA), self.mds.process_payload(node_deleted_payload))
         # Do nothing in case it is unknown/abstract payload type, like ChannelNode
         self.assertEqual((None, NO_ACTION), self.mds.process_payload(node_payload))
@@ -182,6 +181,16 @@ def get_payloads(entity_class):
         self.assertEqual(UNKNOWN_CHANNEL, result[1])
         self.assertEqual(node_dict['metadata_type'], result[0].to_dict()['metadata_type'])
 
+    @db_session
+    def test_process_payload_reject_old(self):
+        # Check there is no action if the processed payload has a timestamp that is less than the
+        # local_version of the corresponding local channel. (I.e. remote peer trying to push back a deleted entry)
+        channel = self.mds.ChannelMetadata(title='bla', version=123, local_version=12)
+        torrent = self.mds.TorrentMetadata(title='blabla', timestamp=11, origin_id=channel.id_)
+        payload = torrent._payload_class(**torrent.to_dict())
+        torrent.delete()
+        self.assertEqual((None, NO_ACTION), self.mds.process_payload(payload))
+
     @db_session
     def test_get_num_channels_nodes(self):
         self.mds.ChannelMetadata(title='testchan', id_=0)

Tribler/Test/Core/Modules/RestApi/test_downloads_endpoint.py

@@ -579,7 +579,7 @@ def verify_download(_):
             self.assertGreaterEqual(len(self.session.get_downloads()), 1)
 
         post_data = {'uri': 'file:%s' % os.path.join(TESTS_DIR, 'Core/data/sample_channel/channel.mdblob')}
-        expected_json = {'started': True, 'infohash': '6853d25535a1c7593e716dd6a69fc3dd7a7bfcc8'}
+        expected_json = {'started': True, 'infohash': '8e1cfb5b832e124b681497578c3715b63df01b50'}
         return self.do_request('downloads', expected_code=200, request_type='PUT', post_data=post_data,
                                expected_json=expected_json).addCallback(verify_download)
 

Tribler/Test/Core/data/sample_channel/channel.torrent

@@ -1 +1 @@
-d13:creation datei1548884024e4:infod5:filesld6:lengthi538e4:pathl23:000000000006.mdblob.lz4eed6:lengthi283e4:pathl23:000000000002.mdblob.lz4eed6:lengthi211e4:pathl23:000000000007.mdblob.lz4eee4:name32:893e876d3d09f0bb87bf95036b3d5e2612:piece lengthi16384e6:pieces20:� �'���kٷ�{�{��75ee
+d13:creation datei1551110113e4:infod5:filesld6:lengthi544e4:pathl24:1551110113006.mdblob.lz4eed6:lengthi211e4:pathl24:1551110113007.mdblob.lz4eed6:lengthi286e4:pathl24:1551110113002.mdblob.lz4eee4:name32:58b2c08b9ccd3e03a29848b3f363832012:piece lengthi16384e6:pieces20:'b���g�a�bJ'�zY�a6iee

Tribler/Test/Core/data/sample_channel/channel_upd.torrent

@@ -1 +1 @@
-d13:creation datei1548884024e4:infod5:filesld6:lengthi538e4:pathl23:000000000006.mdblob.lz4eed6:lengthi283e4:pathl23:000000000002.mdblob.lz4eed6:lengthi223e4:pathl23:000000000009.mdblob.lz4eed6:lengthi211e4:pathl23:000000000007.mdblob.lz4eee4:name32:893e876d3d09f0bb87bf95036b3d5e2612:piece lengthi16384e6:pieces20:E���SzN�I"�����
�ee
+d13:creation datei1551110113e4:infod5:filesld6:lengthi544e4:pathl24:1551110113006.mdblob.lz4eed6:lengthi226e4:pathl24:1551110113009.mdblob.lz4eed6:lengthi211e4:pathl24:1551110113007.mdblob.lz4eed6:lengthi286e4:pathl24:1551110113002.mdblob.lz4eee4:name32:58b2c08b9ccd3e03a29848b3f363832012:piece lengthi16384e6:pieces20:�"8��6;G���H�FQѮMee