GitLab

Philipp Hörist pushed to branch master at gajim / gajim

Commits:

46926e71

by Philipp Hörist at 2017-11-07T20:39:36+01:00

Remove old message duplicate code

- Calculating so many hashes for each Message is quite expensive
- It hides our own implementation bugs, like when we retrieve history
from a MUC with wrong timestamps, or on rejoin. We never know about it
because the Messages are dropped.
- It should not be necessary anymore. The original problem was a bug
in nbxmpp which triggered mass resending of old messages.

7f1a839e

by Philipp Hörist at 2017-11-07T20:39:36+01:00

Deduplicate GC and MAM Messages based on stanza-id

- Add methods to determine the unique/stable id
- Write the id to the stanza_id DB field
- Add method to deduplicate based on the unique/stable id

d4fdf0d7
by Philipp Hörist at 2017-11-07T20:39:45+01:00
```
Shorten search intervall for MAM messages
```

d381179d

by Philipp Hörist at 2017-11-07T21:03:01+01:00

Merge branch 'deduplication' into 'master'

Deduplicate based on stanza-id

See merge request gajim/gajim!148

5 changed files:

gajim/common/connection_handlers.py
gajim/common/connection_handlers_events.py
gajim/common/logger.py
gajim/common/message_archiving.py
gajim/session.py

Changes:

gajim/common/connection_handlers.py

@@ -725,8 +725,6 @@ class ConnectionHandlersBase:
          # IDs of sent messages (https://trac.gajim.org/ticket/8222)
          self.sent_message_ids = []
 -        self.received_message_hashes = []
+-
          # We decrypt GPG messages one after the other. Keep queue in mem
          self.gpg_messages_to_decrypt = []
@@ -946,7 +944,8 @@ class ConnectionHandlersBase:
      def _on_message_received(self, obj):
          if isinstance(obj, MessageReceivedEvent):
              app.nec.push_incoming_event(
 -                DecryptedMessageReceivedEvent(None, conn=self, msg_obj=obj))
 +                DecryptedMessageReceivedEvent(
 +                    None, conn=self, msg_obj=obj, stanza_id=obj.unique_id))
          else:
              app.nec.push_incoming_event(
                  MamDecryptedMessageReceivedEvent(None, **vars(obj)))
@@ -1012,14 +1011,14 @@ class ConnectionHandlersBase:
              return True
          elif obj.mtype == 'groupchat':
              app.nec.push_incoming_event(GcMessageReceivedEvent(None,
 -                conn=self, msg_obj=obj))
 +                conn=self, msg_obj=obj, stanza_id=obj.unique_id))
              return True
      def _nec_gc_message_received(self, obj):
          if obj.conn.name != self.name:
              return
          if (app.config.should_log(obj.conn.name, obj.jid) and
 -            obj.msgtxt and obj.nick):
 +                obj.msgtxt and obj.nick):
              # if not obj.nick, it means message comes from room itself
              # usually it hold description and can be send at each connection
              # so don't store it in logs
@@ -1028,7 +1027,8 @@ class ConnectionHandlersBase:
                                          KindConstant.GC_MSG,
                                          message=obj.msgtxt,
                                          contact_name=obj.nick,
 -                                        additional_data=obj.additional_data)
 +                                        additional_data=obj.additional_data,
 +                                        stanza_id=obj.unique_id)
              app.logger.set_room_last_message_time(obj.room_jid, obj.timestamp)
      # process and dispatch an error message
@@ -2048,8 +2048,6 @@ ConnectionHandlersBase, ConnectionJingle, ConnectionIBBytestream):
          app.nec.push_incoming_event(SignedInEvent(None, conn=self))
          self.send_awaiting_pep()
          self.continue_connect_info = None
 -        # hashes of already received messages
 -        self.received_message_hashes = []
      def _SearchCB(self, con, iq_obj):
          log.debug('SearchCB')

gajim/common/connection_handlers_events.py

@@ -139,6 +139,28 @@ class HelperEvent:
              if oob_desc is not None:
                  self.additional_data['gajim']['oob_desc'] = oob_desc
 +    def get_stanza_id(self, stanza, query=False):
 +        if query:
 +            # On a MAM query the stanza-id is maybe not set, so
 +            # get the id of the stanza
 +            return stanza.getAttr('id')
 +        stanza_id, by = stanza.getStanzaIDAttrs()
 +        if by is None:
 +            # We can not verify who set this stanza-id, ignore it.
 +            return
 +        elif not self.conn.get_own_jid().bareMatch(by):
 +            # by attribute does not match the server, ignore it.
 +            return
 +        return stanza_id
++
 +    @staticmethod
 +    def get_forwarded_message(stanza):
 +        forwarded = stanza.getTag('forwarded',
 +                                  namespace=nbxmpp.NS_FORWARD,
 +                                  protocol=True)
 +        if forwarded is not None:
 +            return forwarded.getTag('message', protocol=True)
++
  class HttpAuthReceivedEvent(nec.NetworkIncomingEvent):
      name = 'http-auth-received'
      base_network_events = []
@@ -1026,6 +1048,7 @@ class MamMessageReceivedEvent(nec.NetworkIncomingEvent, HelperEvent):
          :stanza:        Complete stanza Node
          :forwarded:     Forwarded Node
          :result:        Result Node
 +        :unique_id:     The unique stable id
          '''
          self._set_base_event_vars_as_attributes(base_event)
          self.additional_data = {}
@@ -1046,6 +1069,13 @@ class MamMessageReceivedEvent(nec.NetworkIncomingEvent, HelperEvent):
              log.info('Received groupchat message from user archive')
              return False
 +        # use stanza-id as unique-id
 +        self.unique_id, origin_id = self.get_unique_id()
++
 +        # Check for duplicates
 +        if app.logger.find_stanza_id(self.unique_id, origin_id):
 +            return
++
          self.msgtxt = self.msg_.getTagData('body')
          self.query_id = self.result.getAttr('queryid')
@@ -1057,24 +1087,12 @@ class MamMessageReceivedEvent(nec.NetworkIncomingEvent, HelperEvent):
              to = own_jid
          if frm.bareMatch(own_jid):
 -            self.stanza_id = self.msg_.getOriginID()
 -            if self.stanza_id is None:
 -                self.stanza_id = self.msg_.getID()
+-
              self.with_ = to
              self.kind = KindConstant.CHAT_MSG_SENT
          else:
 -            if self.result.getNamespace() == nbxmpp.NS_MAM_2:
 -                self.stanza_id = self.result.getID()
 -            else:
 -                self.stanza_id = self.msg_.getID()
+-
              self.with_ = frm
              self.kind = KindConstant.CHAT_MSG_RECV
 -        if self.stanza_id is None:
 -            log.debug('Could not retrieve stanza-id')
+-
          delay = self.forwarded.getTagAttr(
              'delay', 'stamp', namespace=nbxmpp.NS_DELAY2)
          if delay is None:
@@ -1097,9 +1115,31 @@ class MamMessageReceivedEvent(nec.NetworkIncomingEvent, HelperEvent):
                  log.warning('Received MAM message with '
                              'invalid user timestamp: %s', user_delay)
 -        log.debug('Received mam-message: stanza id: %s', self.stanza_id)
 +        log.debug('Received mam-message: unique id: %s', self.unique_id)
          return True
 +    def get_unique_id(self):
 +        if self.conn.get_own_jid().bareMatch(self.msg_.getFrom()):
 +            # On our own Messages we have to check for both
 +            # stanza-id and origin-id, because other resources
 +            # maybe not support origin-id
 +            stanza_id = None
 +            if self.result.getNamespace() == nbxmpp.NS_MAM_2:
 +                # Only mam:2 ensures valid stanza-id
 +                stanza_id = self.get_stanza_id(self.result, query=True)
++
 +            # try always to get origin-id because its a message
 +            # we sent.
 +            origin_id = self.msg_.getOriginID()
++
 +            return stanza_id, origin_id
++
 +        # A message we received
 +        elif self.result.getNamespace() == nbxmpp.NS_MAM_2:
 +            # Only mam:2 ensures valid stanza-id
 +            return self.get_stanza_id(self.result, query=True), None
 +        return None, None
++
  class MamDecryptedMessageReceivedEvent(nec.NetworkIncomingEvent, HelperEvent):
      name = 'mam-decrypted-message-received'
      base_network_events = []
@@ -1177,6 +1217,14 @@ class MessageReceivedEvent(nec.NetworkIncomingEvent, HelperEvent):
                          self.stanza.getFrom())
              return
 +        # Check for duplicates
 +        self.unique_id = self.get_unique_id()
 +        # Check groupchat messages for duplicates,
 +        # We do this because of MUC History messages
 +        if self.stanza.getType() == 'groupchat':
 +            if app.logger.find_stanza_id(self.unique_id):
 +                return
++
          address_tag = self.stanza.getTag('addresses',
              namespace=nbxmpp.NS_ADDRESS)
          # Be sure it comes from one of our resource, else ignore address element
@@ -1248,7 +1296,8 @@ class MessageReceivedEvent(nec.NetworkIncomingEvent, HelperEvent):
                               conn=self.conn,
                               stanza=self.stanza,
                               forwarded=forwarded,
 -                             result=result))
 +                             result=result,
 +                             stanza_id=self.unique_id))
              return
          # Mediated invitation?
@@ -1320,6 +1369,38 @@ class MessageReceivedEvent(nec.NetworkIncomingEvent, HelperEvent):
          return True
 +    def get_unique_id(self):
 +        if self.stanza.getType() == 'groupchat':
 +            # TODO: Disco the MUC check if 'urn:xmpp:mam:2' is announced
 +            return self.get_stanza_id(self.stanza)
++
 +        elif self.stanza.getType() != 'chat':
 +            return
++
 +        # Messages we receive live
 +        if self.conn.archiving_namespace != nbxmpp.NS_MAM_2:
 +            # Only mam:2 ensures valid stanza-id
 +            return
++
 +        # Sent Carbon
 +        sent_carbon = self.stanza.getTag('sent',
 +                                         namespace=nbxmpp.NS_CARBONS,
 +                                         protocol=True)
 +        if sent_carbon is not None:
 +            message = self.get_forwarded_message(sent_carbon)
 +            return self.get_stanza_id(message)
++
 +        # Received Carbon
 +        received_carbon = self.stanza.getTag('received',
 +                                             namespace=nbxmpp.NS_CARBONS,
 +                                             protocol=True)
 +        if received_carbon is not None:
 +            message = self.get_forwarded_message(received_carbon)
 +            return self.get_stanza_id(message)
++
 +        # Normal Message
 +        return self.get_stanza_id(self.stanza)
++
  class ZeroconfMessageReceivedEvent(MessageReceivedEvent):
      name = 'message-received'
      base_network_events = []
@@ -1416,6 +1497,7 @@ class DecryptedMessageReceivedEvent(nec.NetworkIncomingEvent, HelperEvent):
          self.stanza = self.msg_obj.stanza
          self.additional_data = self.msg_obj.additional_data
          self.id_ = self.msg_obj.id_
 +        self.unique_id = self.msg_obj.unique_id
          self.jid = self.msg_obj.jid
          self.fjid = self.msg_obj.fjid
          self.resource = self.msg_obj.resource
@@ -1472,22 +1554,6 @@ class DecryptedMessageReceivedEvent(nec.NetworkIncomingEvent, HelperEvent):
          if replace:
              self.correct_id = replace.getAttr('id')
 -        # ignore message duplicates
 -        if self.msgtxt and self.id_ and self.jid:
 -            self.msghash = hashlib.sha256(("%s|%s|%s" % (
 -                hashlib.sha256(self.msgtxt.encode('utf-8')).hexdigest(),
 -                hashlib.sha256(self.id_.encode('utf-8')).hexdigest(),
 -                hashlib.sha256(self.jid.encode('utf-8')).hexdigest())).encode(
 -                'utf-8')).digest()
 -            if self.msghash in self.conn.received_message_hashes:
 -                log.info("Ignoring duplicated message from '%s' with id '%s'" % (self.jid, self.id_))
 -                return False
 -            else:
 -                log.debug("subhashes: msgtxt, id_, jid = ('%s', '%s', '%s')" % (hashlib.sha256(self.msgtxt.encode('utf-8')).hexdigest(), hashlib.sha256(self.id_.encode('utf-8')).hexdigest(), hashlib.sha256(self.jid.encode('utf-8')).hexdigest()))
 -                self.conn.received_message_hashes.append(self.msghash)
 -                # only record the last 20000 hashes (should be about 1MB [32 bytes per hash]
 -                # and about 24 hours if you receive a message every 5 seconds)
 -                self.conn.received_message_hashes = self.conn.received_message_hashes[-20000:]
          return True
  class ChatstateReceivedEvent(nec.NetworkIncomingEvent):
@@ -1513,6 +1579,7 @@ class GcMessageReceivedEvent(nec.NetworkIncomingEvent):
          else:
              self.additional_data = self.msg_obj.additional_data
          self.id_ = self.msg_obj.stanza.getID()
 +        self.unique_id = self.msg_obj.unique_id
          self.fjid = self.msg_obj.fjid
          self.msgtxt = self.msg_obj.msgtxt
          self.jid = self.msg_obj.jid

gajim/common/logger.py

@@ -1064,9 +1064,9 @@ class Logger:
          :param msg:         The message text
          """
 -        # Add 5 minutes around the timestamp
 -        start_time = timestamp - 300
 -        end_time = timestamp + 300
 +        # Add 10 seconds around the timestamp
 +        start_time = timestamp - 10
 +        end_time = timestamp + 10
          log.debug('start: %s, end: %s, jid: %s, message: %s',
                    start_time, end_time, jid, msg)
@@ -1084,6 +1084,33 @@ class Logger:
              return True
          return False
 +    def find_stanza_id(self, stanza_id, origin_id=None):
 +        """
 +        Checks if a stanza-id is already in the `logs` table
++
 +        :param stanza_id:   The stanza-id
++
 +        return True if the stanza-id was found
 +        """
 +        ids = []
 +        if stanza_id is not None:
 +            ids.append(stanza_id)
 +        if origin_id is not None:
 +            ids.append(origin_id)
++
 +        sql = '''
 +              SELECT stanza_id FROM logs
 +              WHERE stanza_id IN ({values}) LIMIT 1
 +              '''.format(values=', '.join('?' * len(ids)))
++
 +        result = self.con.execute(sql, tuple(ids)).fetchone()
++
 +        if result is not None:
 +            log.info('Found duplicated message, stanza-id: %s, origin-id: %s',
 +                     stanza_id, origin_id)
 +            return True
 +        return False
++
      def insert_jid(self, jid, kind=None, type_=JIDConstant.NORMAL_TYPE):
          """
          Insert a new jid into the `jids` table.
@@ -1129,6 +1156,9 @@ class Logger:
          lastrowid = self.con.execute(sql, (jid_id, time_, kind) + tuple(kwargs.values())).lastrowid
 +        log.info('Insert into DB: jid: %s, time: %s, kind: %s, stanza_id: %s',
 +                 jid, time_, kind, kwargs.get('stanza_id', None))
++
          if unread and kind == KindConstant.CHAT_MSG_RECV:
              sql = '''INSERT INTO unread_messages (message_id, jid_id)
                       VALUES (?, (SELECT jid_id FROM jids WHERE jid = ?))'''

gajim/common/message_archiving.py

@@ -127,14 +127,18 @@ class ConnectionArchive313:
      def _nec_mam_decrypted_message_received(self, obj):
          if obj.conn.name != self.name:
              return
 +        # if self.archiving_namespace != nbxmpp.NS_MAM_2:
 +        # Fallback duplicate search without stanza-id
          duplicate = app.logger.search_for_duplicate(
              obj.with_, obj.timestamp, obj.msgtxt)
 -        if not duplicate:
 -            app.logger.insert_into_logs(
 -                obj.with_, obj.timestamp, obj.kind,
 -                unread=False,
 -                message=obj.msgtxt,
 -                additional_data=obj.additional_data)
 +        if duplicate:
 +            return
 +        app.logger.insert_into_logs(
 +            obj.with_, obj.timestamp, obj.kind,
 +            unread=False,
 +            message=obj.msgtxt,
 +            additional_data=obj.additional_data,
 +            stanza_id=obj.unique_id)
      def get_query_id(self):
          self.mam_query_id = self.connection.getAnID()

gajim/session.py

@@ -120,7 +120,8 @@ class ChatControlSession(stanza_session.EncryptedStanzaSession):
                  jid, obj.timestamp, log_type,
                  message=msg_to_log,
                  subject=obj.subject,
 -                additional_data=obj.additional_data)
 +                additional_data=obj.additional_data,
 +                stanza_id=obj.unique_id)
          # Handle chat states
          if contact and (not obj.forwarded or not obj.sent):