From df3a661e4adb7676682a5e3c298a2dfda18b08a1 Mon Sep 17 00:00:00 2001
From: Andrew Morgan <andrew@amorgan.xyz>
Date: Fri, 18 Jan 2019 10:04:47 +0000
Subject: Search for messages across predecessor rooms

Signed-off-by: Andrew Morgan <andrew@amorgan.xyz>
---
 synapse/storage/state.py | 1 +
 1 file changed, 1 insertion(+)

(limited to 'synapse/storage')

diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index a134e9b3e8..49b3ff4a71 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -448,6 +448,7 @@ class StateGroupWorkerStore(EventsWorkerStore, SQLBaseStore):
         Returns:
             deferred: dict of (type, state_key) -> event_id
         """
+
         def _get_current_state_ids_txn(txn):
             txn.execute(
                 """SELECT type, state_key, event_id FROM current_state_events
-- 
cgit 1.4.1


From c9bfb058d85f6205fada062c78a4d1eca119417c Mon Sep 17 00:00:00 2001
From: Andrew Morgan <andrew@amorgan.xyz>
Date: Tue, 22 Jan 2019 11:12:48 +0000
Subject: Fix a bug with single-room search searching all rooms

* Create a new method for getting predecessor rooms
* Remove formatting change
---
 synapse/api/filtering.py   | 15 +++++++++++++--
 synapse/handlers/search.py | 42 ++++++++++--------------------------------
 synapse/storage/state.py   | 29 ++++++++++++++++++++++++++++-
 3 files changed, 51 insertions(+), 35 deletions(-)

(limited to 'synapse/storage')

diff --git a/synapse/api/filtering.py b/synapse/api/filtering.py
index 84000e6422..0d8957175d 100644
--- a/synapse/api/filtering.py
+++ b/synapse/api/filtering.py
@@ -444,8 +444,19 @@ class Filter(object):
     def include_redundant_members(self):
         return self.filter_json.get("include_redundant_members", False)
 
-    def add_room_ids(self, room_ids):
-        self.rooms += room_ids
+    def with_room_ids(self, room_ids):
+        """Returns a new filter with the given room IDs appended.
+
+        Args:
+            room_ids (list): A list of room_ids.
+
+        Returns:
+            filter: A new filter including the given rooms and the old
+                    filter's rooms.
+        """
+        newFilter = self
+        newFilter.rooms += room_ids
+        return newFilter
 
 
 def _matches_wildcard(actual_value, filter_value):
diff --git a/synapse/handlers/search.py b/synapse/handlers/search.py
index 77e7e4e0fb..75c26fe065 100644
--- a/synapse/handlers/search.py
+++ b/synapse/handlers/search.py
@@ -49,39 +49,26 @@ class SearchHandler(BaseHandler):
         The full list of all found rooms in then returned.
 
         Args:
-            room_id (str): The ID of the room to search through.
+            room_id (str): id of the room to search through.
 
         Returns:
-            dict of past room IDs as strings
+            Deferred[iterable[str]]: predecessor room ids
         """
 
         historical_room_ids = []
 
         while True:
-            state_ids = yield self.store.get_current_state_ids(room_id)
-            create_id = state_ids.get((EventTypes.Create, ""))
+            predecessor = yield self.store.get_room_predecessor(room_id)
 
-            # If we can't find the create event, assume we've hit a dead end
-            if not create_id:
-                break
-
-            # Retrieve the room's create event
-            create_event = yield self.store.get_event(create_id)
-
-            if not create_event:
-                break
-
-            # Check if a predecessor room is present
-            predecessor = create_event.content.get("predecessor", None)
+            # If no predecessor, assume we've hit a dead end
             if not predecessor:
                 break
 
             # Add predecessor's room ID
-            historical_room_id = predecessor["room_id"]
-            historical_room_ids.append(historical_room_id)
+            historical_room_ids.append(predecessor["room_id"])
 
             # Scan through the old room for further predecessors
-            room_id = historical_room_id
+            room_id = predecessor["room_id"]
 
         defer.returnValue(historical_room_ids)
 
@@ -185,28 +172,19 @@ class SearchHandler(BaseHandler):
         )
         room_ids = set(r.room_id for r in rooms)
 
-        room_ids = search_filter.filter_rooms(room_ids)
-
         # If doing a subset of all rooms seearch, check if any of the rooms
         # are from an upgraded room, and search their contents as well
-        # XXX: There is the possibility that we don't have a create event for
-        # the room in question, in which case we can't return all the results
-        # we want to.
-        # Ideally we would just return the results we can get now, and
-        # try to get more results from other servers in the background.
         if search_filter.rooms:
             historical_room_ids = []
-            for room_id in room_ids:
+            for room_id in search_filter.rooms:
                 # Add any previous rooms to the search if they exist
                 ids = yield self.get_old_rooms_from_upgraded_room(room_id)
                 historical_room_ids += ids
 
-            # Add any found rooms to the list to search
-            for historical_room_id in historical_room_ids:
-                room_ids.add(historical_room_id)
-
             # Prevent any historical events from being filtered
-            search_filter.add_room_ids(historical_room_ids)
+            search_filter = search_filter.with_room_ids(historical_room_ids)
+
+        room_ids = search_filter.filter_rooms(room_ids)
 
         if batch_group == "room_id":
             room_ids.intersection_update({batch_group_key})
diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index 49b3ff4a71..b064671851 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -437,6 +437,34 @@ class StateGroupWorkerStore(EventsWorkerStore, SQLBaseStore):
         create_event = yield self.get_event(create_id)
         defer.returnValue(create_event.content.get("room_version", "1"))
 
+    @defer.inlineCallbacks
+    def get_room_predecessor(self, room_id):
+        """Get the predecessor room of an upgraded room if one exists.
+        Otherwise return None.
+
+        Args:
+            room_id (str)
+
+        Returns:
+            Deferred[str]: predecessor room id
+        """
+
+        state_ids = yield self.get_current_state_ids(room_id)
+        create_id = state_ids.get((EventTypes.Create, ""))
+
+        # If we can't find the create event, assume we've hit a dead end
+        if not create_id:
+            return None
+
+        # Retrieve the room's create event
+        create_event = yield self.get_event(create_id)
+
+        if not create_event:
+            return None
+
+        # Return predecessor if present
+        return create_event.content.get("predecessor", None)
+
     @cached(max_entries=100000, iterable=True)
     def get_current_state_ids(self, room_id):
         """Get the current state event ids for a room based on the
@@ -448,7 +476,6 @@ class StateGroupWorkerStore(EventsWorkerStore, SQLBaseStore):
         Returns:
             deferred: dict of (type, state_key) -> event_id
         """
-
         def _get_current_state_ids_txn(txn):
             txn.execute(
                 """SELECT type, state_key, event_id FROM current_state_events
-- 
cgit 1.4.1


From c433f6109145f0cf6c80dd07ee118b68a3a0cd4e Mon Sep 17 00:00:00 2001
From: Andrew Morgan <andrew@amorgan.xyz>
Date: Tue, 22 Jan 2019 12:06:36 +0000
Subject: Ensure new filter is actually created

---
 synapse/api/filtering.py | 2 +-
 synapse/storage/state.py | 1 -
 2 files changed, 1 insertion(+), 2 deletions(-)

(limited to 'synapse/storage')

diff --git a/synapse/api/filtering.py b/synapse/api/filtering.py
index 0d8957175d..f3a056110f 100644
--- a/synapse/api/filtering.py
+++ b/synapse/api/filtering.py
@@ -454,7 +454,7 @@ class Filter(object):
             filter: A new filter including the given rooms and the old
                     filter's rooms.
         """
-        newFilter = self
+        newFilter = Filter(self.filter_json)
         newFilter.rooms += room_ids
         return newFilter
 
diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index b064671851..981d1e3600 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -448,7 +448,6 @@ class StateGroupWorkerStore(EventsWorkerStore, SQLBaseStore):
         Returns:
             Deferred[str]: predecessor room id
         """
-
         state_ids = yield self.get_current_state_ids(room_id)
         create_id = state_ids.get((EventTypes.Create, ""))
 
-- 
cgit 1.4.1


From 277e50462d1422ac8cfe2df7cd2213288f3d14c5 Mon Sep 17 00:00:00 2001
From: Andrew Morgan <andrew@amorgan.xyz>
Date: Tue, 22 Jan 2019 12:40:26 +0000
Subject: Do not return in a deferred function

---
 synapse/storage/state.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

(limited to 'synapse/storage')

diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index 981d1e3600..fceb9744aa 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -453,16 +453,16 @@ class StateGroupWorkerStore(EventsWorkerStore, SQLBaseStore):
 
         # If we can't find the create event, assume we've hit a dead end
         if not create_id:
-            return None
+            defer.returnValue(None)
 
         # Retrieve the room's create event
         create_event = yield self.get_event(create_id)
 
         if not create_event:
-            return None
+            defer.returnValue(None)
 
         # Return predecessor if present
-        return create_event.content.get("predecessor", None)
+        defer.returnValue(create_event.content.get("predecessor", None))
 
     @cached(max_entries=100000, iterable=True)
     def get_current_state_ids(self, room_id):
-- 
cgit 1.4.1


From 2557531f0f60ee1891a74babed54800cb1bcfd06 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Tue, 22 Jan 2019 11:55:53 +0000
Subject: Fix bug when removing duplicate rows from user_ips

This was caused by accidentally overwritting a `last_seen` variable
in a for loop, causing the wrong value to be written to the progress
table. The result of which was that we didn't scan sections of the table
when searching for duplicates, and so some duplicates did not get
deleted.
---
 synapse/storage/client_ips.py | 33 ++++++++++++++++++++-------------
 1 file changed, 20 insertions(+), 13 deletions(-)

(limited to 'synapse/storage')

diff --git a/synapse/storage/client_ips.py b/synapse/storage/client_ips.py
index 5d548f250a..9548cfd3f8 100644
--- a/synapse/storage/client_ips.py
+++ b/synapse/storage/client_ips.py
@@ -110,8 +110,13 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
 
     @defer.inlineCallbacks
     def _remove_user_ip_dupes(self, progress, batch_size):
+        # This works function works by scanning the user_ips table in batches
+        # based on `last_seen`. For each row in a batch it searches the rest of
+        # the table to see if there are any duplicates, if there are then they
+        # are removed and replaced with a suitable row.
 
-        last_seen_progress = progress.get("last_seen", 0)
+        # Fetch the start of the batch
+        begin_last_seen = progress.get("last_seen", 0)
 
         def get_last_seen(txn):
             txn.execute(
@@ -122,17 +127,20 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
                 LIMIT 1
                 OFFSET ?
                 """,
-                (last_seen_progress, batch_size)
+                (begin_last_seen, batch_size)
             )
-            results = txn.fetchone()
-            return results
-
-        # Get a last seen that's sufficiently far away enough from the last one
-        last_seen = yield self.runInteraction(
+            row = txn.fetchone()
+            if row:
+                return row[0]
+            else:
+                return None
+
+        # Get a last seen that has roughly `batch_size` since `begin_last_seen`
+        end_last_seen = yield self.runInteraction(
             "user_ips_dups_get_last_seen", get_last_seen
         )
 
-        if not last_seen:
+        if end_last_seen is None:
             # If we get a None then we're reaching the end and just need to
             # delete the last batch.
             last = True
@@ -142,9 +150,8 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
             last_seen = int(self.clock.time_msec()) * 2
         else:
             last = False
-            last_seen = last_seen[0]
 
-        def remove(txn, last_seen_progress, last_seen):
+        def remove(txn, begin_last_seen, end_last_seen):
             # This works by looking at all entries in the given time span, and
             # then for each (user_id, access_token, ip) tuple in that range
             # checking for any duplicates in the rest of the table (via a join).
@@ -166,7 +173,7 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
                 INNER JOIN user_ips USING (user_id, access_token, ip)
                 GROUP BY user_id, access_token, ip
                 HAVING count(*) > 1""",
-                (last_seen_progress, last_seen)
+                (begin_last_seen, end_last_seen)
             )
             res = txn.fetchall()
 
@@ -194,11 +201,11 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
                 )
 
             self._background_update_progress_txn(
-                txn, "user_ips_remove_dupes", {"last_seen": last_seen}
+                txn, "user_ips_remove_dupes", {"last_seen": end_last_seen}
             )
 
         yield self.runInteraction(
-            "user_ips_dups_remove", remove, last_seen_progress, last_seen
+            "user_ips_dups_remove", remove, begin_last_seen, end_last_seen
         )
         if last:
             yield self._end_background_update("user_ips_remove_dupes")
-- 
cgit 1.4.1


From 1c9704f8ab72047a83c6a3b364f3693b332434f2 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Tue, 22 Jan 2019 16:20:33 +0000
Subject: Don't shadow params

---
 synapse/storage/client_ips.py | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

(limited to 'synapse/storage')

diff --git a/synapse/storage/client_ips.py b/synapse/storage/client_ips.py
index 9548cfd3f8..c485211175 100644
--- a/synapse/storage/client_ips.py
+++ b/synapse/storage/client_ips.py
@@ -151,7 +151,7 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
         else:
             last = False
 
-        def remove(txn, begin_last_seen, end_last_seen):
+        def remove(txn):
             # This works by looking at all entries in the given time span, and
             # then for each (user_id, access_token, ip) tuple in that range
             # checking for any duplicates in the rest of the table (via a join).
@@ -204,9 +204,8 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
                 txn, "user_ips_remove_dupes", {"last_seen": end_last_seen}
             )
 
-        yield self.runInteraction(
-            "user_ips_dups_remove", remove, begin_last_seen, end_last_seen
-        )
+        yield self.runInteraction("user_ips_dups_remove", remove)
+
         if last:
             yield self._end_background_update("user_ips_remove_dupes")
 
-- 
cgit 1.4.1


From 7f503f83b92150edeb4cc5ae98f6d9bb2e9cdb49 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Tue, 22 Jan 2019 16:31:05 +0000
Subject: Refactor to rewrite the SQL instead

---
 synapse/storage/client_ips.py | 29 ++++++++++++++++-------------
 1 file changed, 16 insertions(+), 13 deletions(-)

(limited to 'synapse/storage')

diff --git a/synapse/storage/client_ips.py b/synapse/storage/client_ips.py
index c485211175..78721a941a 100644
--- a/synapse/storage/client_ips.py
+++ b/synapse/storage/client_ips.py
@@ -140,16 +140,8 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
             "user_ips_dups_get_last_seen", get_last_seen
         )
 
-        if end_last_seen is None:
-            # If we get a None then we're reaching the end and just need to
-            # delete the last batch.
-            last = True
-
-            # We fake not having an upper bound by using a future date, by
-            # just multiplying the current time by two....
-            last_seen = int(self.clock.time_msec()) * 2
-        else:
-            last = False
+        # If it returns None, then we're processing the last batch
+        last = end_last_seen is None
 
         def remove(txn):
             # This works by looking at all entries in the given time span, and
@@ -160,6 +152,16 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
             # all other duplicates.
             # It is efficient due to the existence of (user_id, access_token,
             # ip) and (last_seen) indices.
+
+            # Define the search space, which requires handling the last batch in
+            # a different way
+            if last:
+                clause = "? <= last_seen"
+                args = (begin_last_seen,)
+            else:
+                clause = "? <= last_seen AND last_seen < ?"
+                args = (begin_last_seen, end_last_seen)
+
             txn.execute(
                 """
                 SELECT user_id, access_token, ip,
@@ -167,13 +169,14 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
                 FROM (
                     SELECT user_id, access_token, ip
                     FROM user_ips
-                    WHERE ? <= last_seen AND last_seen < ?
+                    WHERE {}
                     ORDER BY last_seen
                 ) c
                 INNER JOIN user_ips USING (user_id, access_token, ip)
                 GROUP BY user_id, access_token, ip
-                HAVING count(*) > 1""",
-                (begin_last_seen, end_last_seen)
+                HAVING count(*) > 1
+                """.format(clause),
+                args
             )
             res = txn.fetchall()
 
-- 
cgit 1.4.1


From 90743c9d8910679bb688070b3929f3005e106d00 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erikj@jki.re>
Date: Wed, 23 Jan 2019 08:45:18 +0000
Subject: Fixup removal of duplicate `user_ips` rows (#4432)

* Remove unnecessary ORDER BY clause

* Add logging

* Newsfile
---
 changelog.d/4432.misc         | 1 +
 synapse/storage/client_ips.py | 6 +++++-
 2 files changed, 6 insertions(+), 1 deletion(-)
 create mode 100644 changelog.d/4432.misc

(limited to 'synapse/storage')

diff --git a/changelog.d/4432.misc b/changelog.d/4432.misc
new file mode 100644
index 0000000000..047061ed3c
--- /dev/null
+++ b/changelog.d/4432.misc
@@ -0,0 +1 @@
+Apply a unique index to the user_ips table, preventing duplicates.
diff --git a/synapse/storage/client_ips.py b/synapse/storage/client_ips.py
index 78721a941a..b228a20ac2 100644
--- a/synapse/storage/client_ips.py
+++ b/synapse/storage/client_ips.py
@@ -143,6 +143,11 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
         # If it returns None, then we're processing the last batch
         last = end_last_seen is None
 
+        logger.info(
+            "Scanning for duplicate 'user_ips' rows in range: %s <= last_seen < %s",
+            begin_last_seen, end_last_seen,
+        )
+
         def remove(txn):
             # This works by looking at all entries in the given time span, and
             # then for each (user_id, access_token, ip) tuple in that range
@@ -170,7 +175,6 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
                     SELECT user_id, access_token, ip
                     FROM user_ips
                     WHERE {}
-                    ORDER BY last_seen
                 ) c
                 INNER JOIN user_ips USING (user_id, access_token, ip)
                 GROUP BY user_id, access_token, ip
-- 
cgit 1.4.1


From c5a296b10c53aa523facf16c3a4c918f93b8188c Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Wed, 23 Jan 2019 11:11:52 +0000
Subject: Add support for persisting event format versions

Currently we only have the one event format version defined, but this
adds the necessary infrastructure to persist and fetch the format
versions alongside the events.

We specify the format version rather than the room version as:

1. We don't necessarily know the room version, existing events may be
   either v1 or v2.
2. We'd need to be careful to prevent/handle correctly if different
   events in the same room reported to be of different versions, which
   sounds annoying.
---
 synapse/api/constants.py                              | 13 +++++++++++++
 synapse/events/__init__.py                            |  5 +++++
 synapse/storage/events.py                             |  1 +
 synapse/storage/events_worker.py                      | 19 +++++++++++++++----
 .../storage/schema/delta/53/event_format_version.sql  | 16 ++++++++++++++++
 5 files changed, 50 insertions(+), 4 deletions(-)
 create mode 100644 synapse/storage/schema/delta/53/event_format_version.sql

(limited to 'synapse/storage')

diff --git a/synapse/api/constants.py b/synapse/api/constants.py
index 46c4b4b9dc..51ee078bc3 100644
--- a/synapse/api/constants.py
+++ b/synapse/api/constants.py
@@ -120,6 +120,19 @@ KNOWN_ROOM_VERSIONS = {
     RoomVersions.STATE_V2_TEST,
 }
 
+
+class EventFormatVersions(object):
+    """This is an internal enum for tracking the version of the event format,
+    independently from the room version.
+    """
+    V1 = 1
+
+
+KNOWN_EVENT_FORMAT_VERSIONS = {
+    EventFormatVersions.V1,
+}
+
+
 ServerNoticeMsgType = "m.server_notice"
 ServerNoticeLimitReached = "m.server_notice.usage_limit_reached"
 
diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py
index 84c75495d5..310075c2b1 100644
--- a/synapse/events/__init__.py
+++ b/synapse/events/__init__.py
@@ -18,6 +18,9 @@ from distutils.util import strtobool
 
 import six
 
+from synapse.api.constants import (
+    EventFormatVersions,
+)
 from synapse.util.caches import intern_dict
 from synapse.util.frozenutils import freeze
 
@@ -179,6 +182,8 @@ class EventBase(object):
 
 
 class FrozenEvent(EventBase):
+    format_version = EventFormatVersions.V1  # All events of this type are V1
+
     def __init__(self, event_dict, internal_metadata_dict={}, rejected_reason=None):
         event_dict = dict(event_dict)
 
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 79e0276de6..3e1915fb87 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -1268,6 +1268,7 @@ class EventsStore(StateGroupWorkerStore, EventFederationStore, EventsWorkerStore
                         event.internal_metadata.get_dict()
                     ),
                     "json": encode_json(event_dict(event)),
+                    "format_version": event.format_version,
                 }
                 for event, _ in events_and_contexts
             ],
diff --git a/synapse/storage/events_worker.py b/synapse/storage/events_worker.py
index a8326f5296..8dff91e5f8 100644
--- a/synapse/storage/events_worker.py
+++ b/synapse/storage/events_worker.py
@@ -21,11 +21,11 @@ from canonicaljson import json
 
 from twisted.internet import defer
 
+from synapse.api.constants import EventFormatVersions
 from synapse.api.errors import NotFoundError
 # these are only included to make the type annotations work
-from synapse.events import EventBase  # noqa: F401
-from synapse.events import FrozenEvent
 from synapse.events.snapshot import EventContext  # noqa: F401
+from synapse.events import FrozenEvent
 from synapse.events.utils import prune_event
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.util.logcontext import (
@@ -353,6 +353,7 @@ class EventsWorkerStore(SQLBaseStore):
                     self._get_event_from_row,
                     row["internal_metadata"], row["json"], row["redacts"],
                     rejected_reason=row["rejects"],
+                    format_version=row["format_version"],
                 )
                 for row in rows
             ],
@@ -377,6 +378,7 @@ class EventsWorkerStore(SQLBaseStore):
                 " e.event_id as event_id, "
                 " e.internal_metadata,"
                 " e.json,"
+                " e.format_version, "
                 " r.redacts as redacts,"
                 " rej.event_id as rejects "
                 " FROM event_json as e"
@@ -392,7 +394,7 @@ class EventsWorkerStore(SQLBaseStore):
 
     @defer.inlineCallbacks
     def _get_event_from_row(self, internal_metadata, js, redacted,
-                            rejected_reason=None):
+                            format_version, rejected_reason=None):
         with Measure(self._clock, "_get_event_from_row"):
             d = json.loads(js)
             internal_metadata = json.loads(internal_metadata)
@@ -405,8 +407,17 @@ class EventsWorkerStore(SQLBaseStore):
                     desc="_get_event_from_row_rejected_reason",
                 )
 
+            if format_version is None:
+                # This means that we stored the event before we had the concept
+                # of a event format version, so it must be a V1 event.
+                format_version = EventFormatVersions.V1
+
+            # TODO: When we implement new event formats we'll need to use a
+            # different event python type
+            assert format_version == EventFormatVersions.V1
+
             original_ev = FrozenEvent(
-                d,
+                event_dict=d,
                 internal_metadata_dict=internal_metadata,
                 rejected_reason=rejected_reason,
             )
diff --git a/synapse/storage/schema/delta/53/event_format_version.sql b/synapse/storage/schema/delta/53/event_format_version.sql
new file mode 100644
index 0000000000..1d977c2834
--- /dev/null
+++ b/synapse/storage/schema/delta/53/event_format_version.sql
@@ -0,0 +1,16 @@
+/* Copyright 2019 New Vector Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+ALTER TABLE event_json ADD COLUMN format_version INTEGER;
-- 
cgit 1.4.1


From be1065af598e46224aee2eb7b8bcd7e723a72426 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Wed, 23 Jan 2019 11:48:16 +0000
Subject: isort

---
 synapse/events/__init__.py       | 4 +---
 synapse/storage/events_worker.py | 2 +-
 2 files changed, 2 insertions(+), 4 deletions(-)

(limited to 'synapse/storage')

diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py
index 310075c2b1..9dd6940385 100644
--- a/synapse/events/__init__.py
+++ b/synapse/events/__init__.py
@@ -18,9 +18,7 @@ from distutils.util import strtobool
 
 import six
 
-from synapse.api.constants import (
-    EventFormatVersions,
-)
+from synapse.api.constants import EventFormatVersions
 from synapse.util.caches import intern_dict
 from synapse.util.frozenutils import freeze
 
diff --git a/synapse/storage/events_worker.py b/synapse/storage/events_worker.py
index 8dff91e5f8..599f892858 100644
--- a/synapse/storage/events_worker.py
+++ b/synapse/storage/events_worker.py
@@ -23,9 +23,9 @@ from twisted.internet import defer
 
 from synapse.api.constants import EventFormatVersions
 from synapse.api.errors import NotFoundError
+from synapse.events import FrozenEvent
 # these are only included to make the type annotations work
 from synapse.events.snapshot import EventContext  # noqa: F401
-from synapse.events import FrozenEvent
 from synapse.events.utils import prune_event
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.util.logcontext import (
-- 
cgit 1.4.1


From 886e5acc762b879b606773b511ff92345aef14c6 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Wed, 16 Jan 2019 15:13:07 +0000
Subject: Store rejected remote invite events as outliers

Currently they're stored as non-outliers even though the server isn't in
the room, which can be problematic in places where the code assumes it
has the state for all non outlier events.

In particular, there is an edge case where persisting the leave event
triggers a state resolution, which requires looking up the room version
from state. Since the server doesn't have the state, this causes an
exception to be thrown.
---
 synapse/federation/federation_client.py | 10 ++++++--
 synapse/handlers/federation.py          | 44 +++++++++------------------------
 synapse/storage/roommember.py           |  5 +---
 3 files changed, 21 insertions(+), 38 deletions(-)

(limited to 'synapse/storage')

diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py
index d05ed91d64..8fa726759e 100644
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -32,7 +32,6 @@ from synapse.api.errors import (
     HttpResponseException,
     SynapseError,
 )
-from synapse.events import builder
 from synapse.federation.federation_base import FederationBase, event_from_pdu_json
 from synapse.util import logcontext, unwrapFirstError
 from synapse.util.caches.expiringcache import ExpiringCache
@@ -66,6 +65,8 @@ class FederationClient(FederationBase):
         self.state = hs.get_state_handler()
         self.transport_layer = hs.get_federation_transport_client()
 
+        self.event_builder_factory = hs.get_event_builder_factory()
+
         self._get_pdu_cache = ExpiringCache(
             cache_name="get_pdu_cache",
             clock=self._clock,
@@ -571,7 +572,12 @@ class FederationClient(FederationBase):
             if "prev_state" not in pdu_dict:
                 pdu_dict["prev_state"] = []
 
-            ev = builder.EventBuilder(pdu_dict)
+            # Strip off the fields that we want to clobber.
+            pdu_dict.pop("origin", None)
+            pdu_dict.pop("origin_server_ts", None)
+            pdu_dict.pop("unsigned", None)
+
+            ev = self.event_builder_factory.new(pdu_dict)
 
             defer.returnValue(
                 (destination, ev)
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index a3bb864bb2..70be87cd3d 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -43,10 +43,7 @@ from synapse.api.errors import (
     StoreError,
     SynapseError,
 )
-from synapse.crypto.event_signing import (
-    add_hashes_and_signatures,
-    compute_event_signature,
-)
+from synapse.crypto.event_signing import compute_event_signature
 from synapse.events.validator import EventValidator
 from synapse.replication.http.federation import (
     ReplicationCleanRoomRestServlet,
@@ -58,7 +55,6 @@ from synapse.types import UserID, get_domain_from_id
 from synapse.util import logcontext, unwrapFirstError
 from synapse.util.async_helpers import Linearizer
 from synapse.util.distributor import user_joined_room
-from synapse.util.frozenutils import unfreeze
 from synapse.util.logutils import log_function
 from synapse.util.retryutils import NotRetryingDestination
 from synapse.visibility import filter_events_for_server
@@ -1083,7 +1079,9 @@ class FederationHandler(BaseHandler):
         handled_events = set()
 
         try:
-            event = self._sign_event(event)
+            self._sign_event(event)
+            event.internal_metadata.outlier = False
+
             # Try the host we successfully got a response to /make_join/
             # request first.
             try:
@@ -1289,13 +1287,7 @@ class FederationHandler(BaseHandler):
         event.internal_metadata.outlier = True
         event.internal_metadata.invite_from_remote = True
 
-        event.signatures.update(
-            compute_event_signature(
-                event,
-                self.hs.hostname,
-                self.hs.config.signing_key[0]
-            )
-        )
+        self._sign_event(event)
 
         context = yield self.state_handler.compute_event_context(event)
         yield self.persist_events_and_notify([(event, context)])
@@ -1313,7 +1305,7 @@ class FederationHandler(BaseHandler):
         # Mark as outlier as we don't have any state for this event; we're not
         # even in the room.
         event.internal_metadata.outlier = True
-        event = self._sign_event(event)
+        self._sign_event(event)
 
         # Try the host that we succesfully called /make_leave/ on first for
         # the /send_leave/ request.
@@ -1358,26 +1350,14 @@ class FederationHandler(BaseHandler):
         defer.returnValue((origin, event))
 
     def _sign_event(self, event):
-        event.internal_metadata.outlier = False
-
-        builder = self.event_builder_factory.new(
-            unfreeze(event.get_pdu_json())
-        )
-
-        builder.event_id = self.event_builder_factory.create_event_id()
-        builder.origin = self.hs.hostname
-
-        if not hasattr(event, "signatures"):
-            builder.signatures = {}
-
-        add_hashes_and_signatures(
-            builder,
-            self.hs.hostname,
-            self.hs.config.signing_key[0],
+        event.signatures.update(
+            compute_event_signature(
+                event,
+                self.hs.hostname,
+                self.hs.config.signing_key[0]
+            )
         )
 
-        return builder.build()
-
     @defer.inlineCallbacks
     @log_function
     def on_make_leave_request(self, room_id, user_id):
diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py
index 0707f9a86a..c7488f4259 100644
--- a/synapse/storage/roommember.py
+++ b/synapse/storage/roommember.py
@@ -591,10 +591,7 @@ class RoomMemberStore(RoomMemberWorkerStore):
             # i.e., its something that has just happened.
             # The only current event that can also be an outlier is if its an
             # invite that has come in across federation.
-            is_new_state = not backfilled and (
-                not event.internal_metadata.is_outlier()
-                or event.internal_metadata.is_invite_from_remote()
-            )
+            is_new_state = not backfilled
             is_mine = self.hs.is_mine_id(event.state_key)
             if is_new_state and is_mine:
                 if event.membership == Membership.INVITE:
-- 
cgit 1.4.1


From 7c288c22500e2045d36a29c38d2671fad6484e30 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Wed, 23 Jan 2019 20:05:44 +0000
Subject: Clarify the invite flows

---
 synapse/events/__init__.py     |  8 ++++++--
 synapse/handlers/federation.py | 12 +++++++++++-
 synapse/storage/roommember.py  | 11 +++++++----
 3 files changed, 24 insertions(+), 7 deletions(-)

(limited to 'synapse/storage')

diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py
index 84c75495d5..5030636c7e 100644
--- a/synapse/events/__init__.py
+++ b/synapse/events/__init__.py
@@ -41,8 +41,12 @@ class _EventInternalMetadata(object):
     def is_outlier(self):
         return getattr(self, "outlier", False)
 
-    def is_invite_from_remote(self):
-        return getattr(self, "invite_from_remote", False)
+    def is_new_remote_event(self):
+        """Whether this is a new remote event, like an invite or an invite
+        rejection. This is needed as those events are marked as outliers, but
+        they still need to be processed.
+        """
+        return getattr(self, "new_remote_event", False)
 
     def get_send_on_behalf_of(self):
         """Whether this server should send the event on behalf of another server.
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 9a14ba4517..e017cab777 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -43,6 +43,7 @@ from synapse.api.errors import (
     StoreError,
     SynapseError,
 )
+from synapse.crypto.event_signing import compute_event_signature
 from synapse.events.validator import EventValidator
 from synapse.replication.http.federation import (
     ReplicationCleanRoomRestServlet,
@@ -1283,7 +1284,15 @@ class FederationHandler(BaseHandler):
             )
 
         event.internal_metadata.outlier = True
-        event.internal_metadata.invite_from_remote = True
+        event.internal_metadata.new_remote_event = True
+
+        event.signatures.update(
+            compute_event_signature(
+                event,
+                self.hs.hostname,
+                self.hs.config.signing_key[0]
+            )
+        )
 
         context = yield self.state_handler.compute_event_context(event)
         yield self.persist_events_and_notify([(event, context)])
@@ -1301,6 +1310,7 @@ class FederationHandler(BaseHandler):
         # Mark as outlier as we don't have any state for this event; we're not
         # even in the room.
         event.internal_metadata.outlier = True
+        event.internal_metadata.new_remote_event = True
 
         # Try the host that we succesfully called /make_leave/ on first for
         # the /send_leave/ request.
diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py
index c7488f4259..40b13de80b 100644
--- a/synapse/storage/roommember.py
+++ b/synapse/storage/roommember.py
@@ -588,10 +588,13 @@ class RoomMemberStore(RoomMemberWorkerStore):
             )
 
             # We update the local_invites table only if the event is "current",
-            # i.e., its something that has just happened.
-            # The only current event that can also be an outlier is if its an
-            # invite that has come in across federation.
-            is_new_state = not backfilled
+            # i.e., its something that has just happened. If the event is an
+            # outlier it is only current if its a "new remote event", like a
+            # remote invite or a rejection of a remote invite.
+            is_new_state = not backfilled and (
+                not event.internal_metadata.is_outlier()
+                or event.internal_metadata.is_new_remote_event()
+            )
             is_mine = self.hs.is_mine_id(event.state_key)
             if is_new_state and is_mine:
                 if event.membership == Membership.INVITE:
-- 
cgit 1.4.1


From 58f6c4818337364dd9c6bf01062e7b0dadcb8a25 Mon Sep 17 00:00:00 2001
From: Amber Brown <hawkowl@atleastfornow.net>
Date: Thu, 24 Jan 2019 21:31:54 +1100
Subject: Use native UPSERTs where possible (#4306)

---
 .coveragerc                         |   6 +-
 .gitignore                          |   6 +-
 changelog.d/4306.misc               |   1 +
 synapse/storage/_base.py            | 148 +++++++++++++++++++++++++++++++++---
 synapse/storage/client_ips.py       |   5 +-
 synapse/storage/engines/__init__.py |   2 +-
 synapse/storage/engines/postgres.py |  14 ++++
 synapse/storage/engines/sqlite.py   |  96 +++++++++++++++++++++++
 synapse/storage/engines/sqlite3.py  |  87 ---------------------
 synapse/storage/pusher.py           |   9 ++-
 synapse/storage/user_directory.py   |  55 ++++++++++----
 tests/storage/test_base.py          |   1 +
 tests/test_server.py                |  12 ++-
 tests/unittest.py                   |  12 ++-
 tox.ini                             |   1 +
 15 files changed, 325 insertions(+), 130 deletions(-)
 create mode 100644 changelog.d/4306.misc
 create mode 100644 synapse/storage/engines/sqlite.py
 delete mode 100644 synapse/storage/engines/sqlite3.py

(limited to 'synapse/storage')

diff --git a/.coveragerc b/.coveragerc
index 9873a30738..e9460a340a 100644
--- a/.coveragerc
+++ b/.coveragerc
@@ -1,11 +1,7 @@
 [run]
 branch = True
 parallel = True
-source = synapse
-
-[paths]
-source=
-   coverage
+include = synapse/*
 
 [report]
 precision = 2
diff --git a/.gitignore b/.gitignore
index d739595c3a..1033124f1d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -25,9 +25,9 @@ homeserver*.pid
 *.tls.dh
 *.tls.key
 
-.coverage
-.coverage.*
-!.coverage.rc
+.coverage*
+coverage.*
+!.coveragerc
 htmlcov
 
 demo/*/*.db
diff --git a/changelog.d/4306.misc b/changelog.d/4306.misc
new file mode 100644
index 0000000000..58130b6190
--- /dev/null
+++ b/changelog.d/4306.misc
@@ -0,0 +1 @@
+Synapse will now take advantage of native UPSERT functionality in PostgreSQL 9.5+ and SQLite 3.24+.
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 865b5e915a..254fdc04c6 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -192,6 +192,41 @@ class SQLBaseStore(object):
 
         self.database_engine = hs.database_engine
 
+        # A set of tables that are not safe to use native upserts in.
+        self._unsafe_to_upsert_tables = {"user_ips"}
+
+        if self.database_engine.can_native_upsert:
+            # Check ASAP (and then later, every 1s) to see if we have finished
+            # background updates of tables that aren't safe to update.
+            self._clock.call_later(0.0, self._check_safe_to_upsert)
+
+    @defer.inlineCallbacks
+    def _check_safe_to_upsert(self):
+        """
+        Is it safe to use native UPSERT?
+
+        If there are background updates, we will need to wait, as they may be
+        the addition of indexes that set the UNIQUE constraint that we require.
+
+        If the background updates have not completed, wait a second and check again.
+        """
+        updates = yield self._simple_select_list(
+            "background_updates",
+            keyvalues=None,
+            retcols=["update_name"],
+            desc="check_background_updates",
+        )
+        updates = [x["update_name"] for x in updates]
+
+        # The User IPs table in schema #53 was missing a unique index, which we
+        # run as a background update.
+        if "user_ips_device_unique_index" not in updates:
+            self._unsafe_to_upsert_tables.discard("user_id")
+
+        # If there's any tables left to check, reschedule to run.
+        if self._unsafe_to_upsert_tables:
+            self._clock.call_later(1.0, self._check_safe_to_upsert)
+
     def start_profiling(self):
         self._previous_loop_ts = self._clock.time_msec()
 
@@ -494,8 +529,15 @@ class SQLBaseStore(object):
         txn.executemany(sql, vals)
 
     @defer.inlineCallbacks
-    def _simple_upsert(self, table, keyvalues, values,
-                       insertion_values={}, desc="_simple_upsert", lock=True):
+    def _simple_upsert(
+        self,
+        table,
+        keyvalues,
+        values,
+        insertion_values={},
+        desc="_simple_upsert",
+        lock=True
+    ):
         """
 
         `lock` should generally be set to True (the default), but can be set
@@ -516,16 +558,21 @@ class SQLBaseStore(object):
                 inserting
             lock (bool): True to lock the table when doing the upsert.
         Returns:
-            Deferred(bool): True if a new entry was created, False if an
-                existing one was updated.
+            Deferred(None or bool): Native upserts always return None. Emulated
+            upserts return True if a new entry was created, False if an existing
+            one was updated.
         """
         attempts = 0
         while True:
             try:
                 result = yield self.runInteraction(
                     desc,
-                    self._simple_upsert_txn, table, keyvalues, values, insertion_values,
-                    lock=lock
+                    self._simple_upsert_txn,
+                    table,
+                    keyvalues,
+                    values,
+                    insertion_values,
+                    lock=lock,
                 )
                 defer.returnValue(result)
             except self.database_engine.module.IntegrityError as e:
@@ -537,12 +584,59 @@ class SQLBaseStore(object):
 
                 # presumably we raced with another transaction: let's retry.
                 logger.warn(
-                    "IntegrityError when upserting into %s; retrying: %s",
-                    table, e
+                    "%s when upserting into %s; retrying: %s", e.__name__, table, e
                 )
 
-    def _simple_upsert_txn(self, txn, table, keyvalues, values, insertion_values={},
-                           lock=True):
+    def _simple_upsert_txn(
+        self,
+        txn,
+        table,
+        keyvalues,
+        values,
+        insertion_values={},
+        lock=True,
+    ):
+        """
+        Pick the UPSERT method which works best on the platform. Either the
+        native one (Pg9.5+, recent SQLites), or fall back to an emulated method.
+
+        Args:
+            txn: The transaction to use.
+            table (str): The table to upsert into
+            keyvalues (dict): The unique key tables and their new values
+            values (dict): The nonunique columns and their new values
+            insertion_values (dict): additional key/values to use only when
+                inserting
+            lock (bool): True to lock the table when doing the upsert.
+        Returns:
+            Deferred(None or bool): Native upserts always return None. Emulated
+            upserts return True if a new entry was created, False if an existing
+            one was updated.
+        """
+        if (
+            self.database_engine.can_native_upsert
+            and table not in self._unsafe_to_upsert_tables
+        ):
+            return self._simple_upsert_txn_native_upsert(
+                txn,
+                table,
+                keyvalues,
+                values,
+                insertion_values=insertion_values,
+            )
+        else:
+            return self._simple_upsert_txn_emulated(
+                txn,
+                table,
+                keyvalues,
+                values,
+                insertion_values=insertion_values,
+                lock=lock,
+            )
+
+    def _simple_upsert_txn_emulated(
+        self, txn, table, keyvalues, values, insertion_values={}, lock=True
+    ):
         # We need to lock the table :(, unless we're *really* careful
         if lock:
             self.database_engine.lock_table(txn, table)
@@ -577,12 +671,44 @@ class SQLBaseStore(object):
         sql = "INSERT INTO %s (%s) VALUES (%s)" % (
             table,
             ", ".join(k for k in allvalues),
-            ", ".join("?" for _ in allvalues)
+            ", ".join("?" for _ in allvalues),
         )
         txn.execute(sql, list(allvalues.values()))
         # successfully inserted
         return True
 
+    def _simple_upsert_txn_native_upsert(
+        self, txn, table, keyvalues, values, insertion_values={}
+    ):
+        """
+        Use the native UPSERT functionality in recent PostgreSQL versions.
+
+        Args:
+            table (str): The table to upsert into
+            keyvalues (dict): The unique key tables and their new values
+            values (dict): The nonunique columns and their new values
+            insertion_values (dict): additional key/values to use only when
+                inserting
+        Returns:
+            None
+        """
+        allvalues = {}
+        allvalues.update(keyvalues)
+        allvalues.update(values)
+        allvalues.update(insertion_values)
+
+        sql = (
+            "INSERT INTO %s (%s) VALUES (%s) "
+            "ON CONFLICT (%s) DO UPDATE SET %s"
+        ) % (
+            table,
+            ", ".join(k for k in allvalues),
+            ", ".join("?" for _ in allvalues),
+            ", ".join(k for k in keyvalues),
+            ", ".join(k + "=EXCLUDED." + k for k in values),
+        )
+        txn.execute(sql, list(allvalues.values()))
+
     def _simple_select_one(self, table, keyvalues, retcols,
                            allow_none=False, desc="_simple_select_one"):
         """Executes a SELECT query on the named table, which is expected to
diff --git a/synapse/storage/client_ips.py b/synapse/storage/client_ips.py
index b228a20ac2..091d7116c5 100644
--- a/synapse/storage/client_ips.py
+++ b/synapse/storage/client_ips.py
@@ -257,7 +257,10 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
         )
 
     def _update_client_ips_batch_txn(self, txn, to_update):
-        self.database_engine.lock_table(txn, "user_ips")
+        if "user_ips" in self._unsafe_to_upsert_tables or (
+            not self.database_engine.can_native_upsert
+        ):
+            self.database_engine.lock_table(txn, "user_ips")
 
         for entry in iteritems(to_update):
             (user_id, access_token, ip), (user_agent, device_id, last_seen) = entry
diff --git a/synapse/storage/engines/__init__.py b/synapse/storage/engines/__init__.py
index e2f9de8451..ff5ef97ca8 100644
--- a/synapse/storage/engines/__init__.py
+++ b/synapse/storage/engines/__init__.py
@@ -18,7 +18,7 @@ import platform
 
 from ._base import IncorrectDatabaseSetup
 from .postgres import PostgresEngine
-from .sqlite3 import Sqlite3Engine
+from .sqlite import Sqlite3Engine
 
 SUPPORTED_MODULE = {
     "sqlite3": Sqlite3Engine,
diff --git a/synapse/storage/engines/postgres.py b/synapse/storage/engines/postgres.py
index 42225f8a2a..4004427c7b 100644
--- a/synapse/storage/engines/postgres.py
+++ b/synapse/storage/engines/postgres.py
@@ -38,6 +38,13 @@ class PostgresEngine(object):
         return sql.replace("?", "%s")
 
     def on_new_connection(self, db_conn):
+
+        # Get the version of PostgreSQL that we're using. As per the psycopg2
+        # docs: The number is formed by converting the major, minor, and
+        # revision numbers into two-decimal-digit numbers and appending them
+        # together. For example, version 8.1.5 will be returned as 80105
+        self._version = db_conn.server_version
+
         db_conn.set_isolation_level(
             self.module.extensions.ISOLATION_LEVEL_REPEATABLE_READ
         )
@@ -54,6 +61,13 @@ class PostgresEngine(object):
 
         cursor.close()
 
+    @property
+    def can_native_upsert(self):
+        """
+        Can we use native UPSERTs? This requires PostgreSQL 9.5+.
+        """
+        return self._version >= 90500
+
     def is_deadlock(self, error):
         if isinstance(error, self.module.DatabaseError):
             # https://www.postgresql.org/docs/current/static/errcodes-appendix.html
diff --git a/synapse/storage/engines/sqlite.py b/synapse/storage/engines/sqlite.py
new file mode 100644
index 0000000000..c64d73ff21
--- /dev/null
+++ b/synapse/storage/engines/sqlite.py
@@ -0,0 +1,96 @@
+# -*- coding: utf-8 -*-
+# Copyright 2015, 2016 OpenMarket Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import struct
+import threading
+from sqlite3 import sqlite_version_info
+
+from synapse.storage.prepare_database import prepare_database
+
+
+class Sqlite3Engine(object):
+    single_threaded = True
+
+    def __init__(self, database_module, database_config):
+        self.module = database_module
+
+        # The current max state_group, or None if we haven't looked
+        # in the DB yet.
+        self._current_state_group_id = None
+        self._current_state_group_id_lock = threading.Lock()
+
+    @property
+    def can_native_upsert(self):
+        """
+        Do we support native UPSERTs? This requires SQLite3 3.24+, plus some
+        more work we haven't done yet to tell what was inserted vs updated.
+        """
+        return sqlite_version_info >= (3, 24, 0)
+
+    def check_database(self, txn):
+        pass
+
+    def convert_param_style(self, sql):
+        return sql
+
+    def on_new_connection(self, db_conn):
+        prepare_database(db_conn, self, config=None)
+        db_conn.create_function("rank", 1, _rank)
+
+    def is_deadlock(self, error):
+        return False
+
+    def is_connection_closed(self, conn):
+        return False
+
+    def lock_table(self, txn, table):
+        return
+
+    def get_next_state_group_id(self, txn):
+        """Returns an int that can be used as a new state_group ID
+        """
+        # We do application locking here since if we're using sqlite then
+        # we are a single process synapse.
+        with self._current_state_group_id_lock:
+            if self._current_state_group_id is None:
+                txn.execute("SELECT COALESCE(max(id), 0) FROM state_groups")
+                self._current_state_group_id = txn.fetchone()[0]
+
+            self._current_state_group_id += 1
+            return self._current_state_group_id
+
+
+# Following functions taken from: https://github.com/coleifer/peewee
+
+def _parse_match_info(buf):
+    bufsize = len(buf)
+    return [struct.unpack('@I', buf[i:i + 4])[0] for i in range(0, bufsize, 4)]
+
+
+def _rank(raw_match_info):
+    """Handle match_info called w/default args 'pcx' - based on the example rank
+    function http://sqlite.org/fts3.html#appendix_a
+    """
+    match_info = _parse_match_info(raw_match_info)
+    score = 0.0
+    p, c = match_info[:2]
+    for phrase_num in range(p):
+        phrase_info_idx = 2 + (phrase_num * c * 3)
+        for col_num in range(c):
+            col_idx = phrase_info_idx + (col_num * 3)
+            x1, x2 = match_info[col_idx:col_idx + 2]
+            if x1 > 0:
+                score += float(x1) / x2
+    return score
diff --git a/synapse/storage/engines/sqlite3.py b/synapse/storage/engines/sqlite3.py
deleted file mode 100644
index 19949fc474..0000000000
--- a/synapse/storage/engines/sqlite3.py
+++ /dev/null
@@ -1,87 +0,0 @@
-# -*- coding: utf-8 -*-
-# Copyright 2015, 2016 OpenMarket Ltd
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import struct
-import threading
-
-from synapse.storage.prepare_database import prepare_database
-
-
-class Sqlite3Engine(object):
-    single_threaded = True
-
-    def __init__(self, database_module, database_config):
-        self.module = database_module
-
-        # The current max state_group, or None if we haven't looked
-        # in the DB yet.
-        self._current_state_group_id = None
-        self._current_state_group_id_lock = threading.Lock()
-
-    def check_database(self, txn):
-        pass
-
-    def convert_param_style(self, sql):
-        return sql
-
-    def on_new_connection(self, db_conn):
-        prepare_database(db_conn, self, config=None)
-        db_conn.create_function("rank", 1, _rank)
-
-    def is_deadlock(self, error):
-        return False
-
-    def is_connection_closed(self, conn):
-        return False
-
-    def lock_table(self, txn, table):
-        return
-
-    def get_next_state_group_id(self, txn):
-        """Returns an int that can be used as a new state_group ID
-        """
-        # We do application locking here since if we're using sqlite then
-        # we are a single process synapse.
-        with self._current_state_group_id_lock:
-            if self._current_state_group_id is None:
-                txn.execute("SELECT COALESCE(max(id), 0) FROM state_groups")
-                self._current_state_group_id = txn.fetchone()[0]
-
-            self._current_state_group_id += 1
-            return self._current_state_group_id
-
-
-# Following functions taken from: https://github.com/coleifer/peewee
-
-def _parse_match_info(buf):
-    bufsize = len(buf)
-    return [struct.unpack('@I', buf[i:i + 4])[0] for i in range(0, bufsize, 4)]
-
-
-def _rank(raw_match_info):
-    """Handle match_info called w/default args 'pcx' - based on the example rank
-    function http://sqlite.org/fts3.html#appendix_a
-    """
-    match_info = _parse_match_info(raw_match_info)
-    score = 0.0
-    p, c = match_info[:2]
-    for phrase_num in range(p):
-        phrase_info_idx = 2 + (phrase_num * c * 3)
-        for col_num in range(c):
-            col_idx = phrase_info_idx + (col_num * 3)
-            x1, x2 = match_info[col_idx:col_idx + 2]
-            if x1 > 0:
-                score += float(x1) / x2
-    return score
diff --git a/synapse/storage/pusher.py b/synapse/storage/pusher.py
index 2743b52bad..134297e284 100644
--- a/synapse/storage/pusher.py
+++ b/synapse/storage/pusher.py
@@ -215,7 +215,7 @@ class PusherStore(PusherWorkerStore):
         with self._pushers_id_gen.get_next() as stream_id:
             # no need to lock because `pushers` has a unique key on
             # (app_id, pushkey, user_name) so _simple_upsert will retry
-            newly_inserted = yield self._simple_upsert(
+            yield self._simple_upsert(
                 table="pushers",
                 keyvalues={
                     "app_id": app_id,
@@ -238,7 +238,12 @@ class PusherStore(PusherWorkerStore):
                 lock=False,
             )
 
-            if newly_inserted:
+            user_has_pusher = self.get_if_user_has_pusher.cache.get(
+                (user_id,), None, update_metrics=False
+            )
+
+            if user_has_pusher is not True:
+                # invalidate, since we the user might not have had a pusher before
                 yield self.runInteraction(
                     "add_pusher",
                     self._invalidate_cache_and_stream,
diff --git a/synapse/storage/user_directory.py b/synapse/storage/user_directory.py
index a8781b0e5d..ce48212265 100644
--- a/synapse/storage/user_directory.py
+++ b/synapse/storage/user_directory.py
@@ -168,14 +168,14 @@ class UserDirectoryStore(SQLBaseStore):
             if isinstance(self.database_engine, PostgresEngine):
                 # We weight the localpart most highly, then display name and finally
                 # server name
-                if new_entry:
+                if self.database_engine.can_native_upsert:
                     sql = """
                         INSERT INTO user_directory_search(user_id, vector)
                         VALUES (?,
                             setweight(to_tsvector('english', ?), 'A')
                             || setweight(to_tsvector('english', ?), 'D')
                             || setweight(to_tsvector('english', COALESCE(?, '')), 'B')
-                        )
+                        ) ON CONFLICT (user_id) DO UPDATE SET vector=EXCLUDED.vector
                     """
                     txn.execute(
                         sql,
@@ -185,20 +185,45 @@ class UserDirectoryStore(SQLBaseStore):
                         )
                     )
                 else:
-                    sql = """
-                        UPDATE user_directory_search
-                        SET vector = setweight(to_tsvector('english', ?), 'A')
-                            || setweight(to_tsvector('english', ?), 'D')
-                            || setweight(to_tsvector('english', COALESCE(?, '')), 'B')
-                        WHERE user_id = ?
-                    """
-                    txn.execute(
-                        sql,
-                        (
-                            get_localpart_from_id(user_id), get_domain_from_id(user_id),
-                            display_name, user_id,
+                    # TODO: Remove this code after we've bumped the minimum version
+                    # of postgres to always support upserts, so we can get rid of
+                    # `new_entry` usage
+                    if new_entry is True:
+                        sql = """
+                            INSERT INTO user_directory_search(user_id, vector)
+                            VALUES (?,
+                                setweight(to_tsvector('english', ?), 'A')
+                                || setweight(to_tsvector('english', ?), 'D')
+                                || setweight(to_tsvector('english', COALESCE(?, '')), 'B')
+                            )
+                        """
+                        txn.execute(
+                            sql,
+                            (
+                                user_id, get_localpart_from_id(user_id),
+                                get_domain_from_id(user_id), display_name,
+                            )
+                        )
+                    elif new_entry is False:
+                        sql = """
+                            UPDATE user_directory_search
+                            SET vector = setweight(to_tsvector('english', ?), 'A')
+                                || setweight(to_tsvector('english', ?), 'D')
+                                || setweight(to_tsvector('english', COALESCE(?, '')), 'B')
+                            WHERE user_id = ?
+                        """
+                        txn.execute(
+                            sql,
+                            (
+                                get_localpart_from_id(user_id),
+                                get_domain_from_id(user_id),
+                                display_name, user_id,
+                            )
+                        )
+                    else:
+                        raise RuntimeError(
+                            "upsert returned None when 'can_native_upsert' is False"
                         )
-                    )
             elif isinstance(self.database_engine, Sqlite3Engine):
                 value = "%s %s" % (user_id, display_name,) if display_name else user_id
                 self._simple_upsert_txn(
diff --git a/tests/storage/test_base.py b/tests/storage/test_base.py
index 829f47d2e8..452d76ddd5 100644
--- a/tests/storage/test_base.py
+++ b/tests/storage/test_base.py
@@ -49,6 +49,7 @@ class SQLBaseStoreTestCase(unittest.TestCase):
         self.db_pool.runWithConnection = runWithConnection
 
         config = Mock()
+        config._enable_native_upserts = False
         config.event_cache_size = 1
         config.database_config = {"name": "sqlite3"}
         hs = TestHomeServer(
diff --git a/tests/test_server.py b/tests/test_server.py
index 634a8fbca5..08fb3fe02f 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -19,7 +19,7 @@ from six import StringIO
 
 from twisted.internet.defer import Deferred
 from twisted.python.failure import Failure
-from twisted.test.proto_helpers import AccumulatingProtocol, MemoryReactorClock
+from twisted.test.proto_helpers import AccumulatingProtocol
 from twisted.web.resource import Resource
 from twisted.web.server import NOT_DONE_YET
 
@@ -30,12 +30,18 @@ from synapse.util import Clock
 from synapse.util.logcontext import make_deferred_yieldable
 
 from tests import unittest
-from tests.server import FakeTransport, make_request, render, setup_test_homeserver
+from tests.server import (
+    FakeTransport,
+    ThreadedMemoryReactorClock,
+    make_request,
+    render,
+    setup_test_homeserver,
+)
 
 
 class JsonResourceTests(unittest.TestCase):
     def setUp(self):
-        self.reactor = MemoryReactorClock()
+        self.reactor = ThreadedMemoryReactorClock()
         self.hs_clock = Clock(self.reactor)
         self.homeserver = setup_test_homeserver(
             self.addCleanup, http_client=None, clock=self.hs_clock, reactor=self.reactor
diff --git a/tests/unittest.py b/tests/unittest.py
index 78d2f740f9..cda549c783 100644
--- a/tests/unittest.py
+++ b/tests/unittest.py
@@ -96,7 +96,7 @@ class TestCase(unittest.TestCase):
 
         method = getattr(self, methodName)
 
-        level = getattr(method, "loglevel", getattr(self, "loglevel", logging.ERROR))
+        level = getattr(method, "loglevel", getattr(self, "loglevel", logging.WARNING))
 
         @around(self)
         def setUp(orig):
@@ -333,7 +333,15 @@ class HomeserverTestCase(TestCase):
         """
         kwargs = dict(kwargs)
         kwargs.update(self._hs_args)
-        return setup_test_homeserver(self.addCleanup, *args, **kwargs)
+        hs = setup_test_homeserver(self.addCleanup, *args, **kwargs)
+        stor = hs.get_datastore()
+
+        # Run the database background updates.
+        if hasattr(stor, "do_next_background_update"):
+            while not self.get_success(stor.has_completed_background_updates()):
+                self.get_success(stor.do_next_background_update(1))
+
+        return hs
 
     def pump(self, by=0.0):
         """
diff --git a/tox.ini b/tox.ini
index a0f5486829..9b2d78ed6d 100644
--- a/tox.ini
+++ b/tox.ini
@@ -149,4 +149,5 @@ deps =
     codecov
 commands =
     coverage combine
+    coverage xml
     codecov -X gcov
-- 
cgit 1.4.1


From e8c9f1539716b9593de837343da5dc4fa9bc10b0 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 24 Jan 2019 11:14:07 +0000
Subject: Replace missed usages of FrozenEvent

---
 synapse/replication/http/federation.py | 8 ++++++--
 synapse/replication/http/send_event.py | 8 ++++++--
 synapse/storage/events_worker.py       | 8 ++------
 3 files changed, 14 insertions(+), 10 deletions(-)

(limited to 'synapse/storage')

diff --git a/synapse/replication/http/federation.py b/synapse/replication/http/federation.py
index 64a79da162..2e16c69666 100644
--- a/synapse/replication/http/federation.py
+++ b/synapse/replication/http/federation.py
@@ -17,7 +17,7 @@ import logging
 
 from twisted.internet import defer
 
-from synapse.events import FrozenEvent
+from synapse.events import event_type_from_format_version
 from synapse.events.snapshot import EventContext
 from synapse.http.servlet import parse_json_object_from_request
 from synapse.replication.http._base import ReplicationEndpoint
@@ -70,6 +70,7 @@ class ReplicationFederationSendEventsRestServlet(ReplicationEndpoint):
 
             event_payloads.append({
                 "event": event.get_pdu_json(),
+                "event_format_version": event.format_version,
                 "internal_metadata": event.internal_metadata.get_dict(),
                 "rejected_reason": event.rejected_reason,
                 "context": serialized_context,
@@ -94,9 +95,12 @@ class ReplicationFederationSendEventsRestServlet(ReplicationEndpoint):
             event_and_contexts = []
             for event_payload in event_payloads:
                 event_dict = event_payload["event"]
+                format_ver = content["event_format_version"]
                 internal_metadata = event_payload["internal_metadata"]
                 rejected_reason = event_payload["rejected_reason"]
-                event = FrozenEvent(event_dict, internal_metadata, rejected_reason)
+
+                EventType = event_type_from_format_version(format_ver)
+                event = EventType(event_dict, internal_metadata, rejected_reason)
 
                 context = yield EventContext.deserialize(
                     self.store, event_payload["context"],
diff --git a/synapse/replication/http/send_event.py b/synapse/replication/http/send_event.py
index 5b52c91650..3635015eda 100644
--- a/synapse/replication/http/send_event.py
+++ b/synapse/replication/http/send_event.py
@@ -17,7 +17,7 @@ import logging
 
 from twisted.internet import defer
 
-from synapse.events import FrozenEvent
+from synapse.events import event_type_from_format_version
 from synapse.events.snapshot import EventContext
 from synapse.http.servlet import parse_json_object_from_request
 from synapse.replication.http._base import ReplicationEndpoint
@@ -74,6 +74,7 @@ class ReplicationSendEventRestServlet(ReplicationEndpoint):
 
         payload = {
             "event": event.get_pdu_json(),
+            "event_format_version": event.format_version,
             "internal_metadata": event.internal_metadata.get_dict(),
             "rejected_reason": event.rejected_reason,
             "context": serialized_context,
@@ -90,9 +91,12 @@ class ReplicationSendEventRestServlet(ReplicationEndpoint):
             content = parse_json_object_from_request(request)
 
             event_dict = content["event"]
+            format_ver = content["event_format_version"]
             internal_metadata = content["internal_metadata"]
             rejected_reason = content["rejected_reason"]
-            event = FrozenEvent(event_dict, internal_metadata, rejected_reason)
+
+            EventType = event_type_from_format_version(format_ver)
+            event = EventType(event_dict, internal_metadata, rejected_reason)
 
             requester = Requester.deserialize(self.store, content["requester"])
             context = yield EventContext.deserialize(self.store, content["context"])
diff --git a/synapse/storage/events_worker.py b/synapse/storage/events_worker.py
index 599f892858..0a0ca58fc4 100644
--- a/synapse/storage/events_worker.py
+++ b/synapse/storage/events_worker.py
@@ -23,7 +23,7 @@ from twisted.internet import defer
 
 from synapse.api.constants import EventFormatVersions
 from synapse.api.errors import NotFoundError
-from synapse.events import FrozenEvent
+from synapse.events import FrozenEvent, event_type_from_format_version  # noqa: F401
 # these are only included to make the type annotations work
 from synapse.events.snapshot import EventContext  # noqa: F401
 from synapse.events.utils import prune_event
@@ -412,11 +412,7 @@ class EventsWorkerStore(SQLBaseStore):
                 # of a event format version, so it must be a V1 event.
                 format_version = EventFormatVersions.V1
 
-            # TODO: When we implement new event formats we'll need to use a
-            # different event python type
-            assert format_version == EventFormatVersions.V1
-
-            original_ev = FrozenEvent(
+            original_ev = event_type_from_format_version(format_version)(
                 event_dict=d,
                 internal_metadata_dict=internal_metadata,
                 rejected_reason=rejected_reason,
-- 
cgit 1.4.1


From 0e27501ee5454aca019cdfbfe899027aa8ae44a1 Mon Sep 17 00:00:00 2001
From: Amber Brown <hawkowl@atleastfornow.net>
Date: Thu, 24 Jan 2019 22:57:41 +1100
Subject: Fix UPSERT check (#4459)

---
 changelog.d/4459.misc    |  1 +
 synapse/storage/_base.py | 33 ++++++++++++++++++++++++++++-----
 2 files changed, 29 insertions(+), 5 deletions(-)
 create mode 100644 changelog.d/4459.misc

(limited to 'synapse/storage')

diff --git a/changelog.d/4459.misc b/changelog.d/4459.misc
new file mode 100644
index 0000000000..58130b6190
--- /dev/null
+++ b/changelog.d/4459.misc
@@ -0,0 +1 @@
+Synapse will now take advantage of native UPSERT functionality in PostgreSQL 9.5+ and SQLite 3.24+.
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 254fdc04c6..f62f70b9f1 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -26,6 +26,7 @@ from prometheus_client import Histogram
 from twisted.internet import defer
 
 from synapse.api.errors import StoreError
+from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.storage.engines import PostgresEngine
 from synapse.util.caches.descriptors import Cache
 from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
@@ -198,7 +199,12 @@ class SQLBaseStore(object):
         if self.database_engine.can_native_upsert:
             # Check ASAP (and then later, every 1s) to see if we have finished
             # background updates of tables that aren't safe to update.
-            self._clock.call_later(0.0, self._check_safe_to_upsert)
+            self._clock.call_later(
+                0.0,
+                run_as_background_process,
+                "upsert_safety_check",
+                self._check_safe_to_upsert
+            )
 
     @defer.inlineCallbacks
     def _check_safe_to_upsert(self):
@@ -208,7 +214,7 @@ class SQLBaseStore(object):
         If there are background updates, we will need to wait, as they may be
         the addition of indexes that set the UNIQUE constraint that we require.
 
-        If the background updates have not completed, wait a second and check again.
+        If the background updates have not completed, wait 15 sec and check again.
         """
         updates = yield self._simple_select_list(
             "background_updates",
@@ -221,11 +227,16 @@ class SQLBaseStore(object):
         # The User IPs table in schema #53 was missing a unique index, which we
         # run as a background update.
         if "user_ips_device_unique_index" not in updates:
-            self._unsafe_to_upsert_tables.discard("user_id")
+            self._unsafe_to_upsert_tables.discard("user_ips")
 
         # If there's any tables left to check, reschedule to run.
         if self._unsafe_to_upsert_tables:
-            self._clock.call_later(1.0, self._check_safe_to_upsert)
+            self._clock.call_later(
+                15.0,
+                run_as_background_process,
+                "upsert_safety_check",
+                self._check_safe_to_upsert
+            )
 
     def start_profiling(self):
         self._previous_loop_ts = self._clock.time_msec()
@@ -609,7 +620,7 @@ class SQLBaseStore(object):
                 inserting
             lock (bool): True to lock the table when doing the upsert.
         Returns:
-            Deferred(None or bool): Native upserts always return None. Emulated
+            None or bool: Native upserts always return None. Emulated
             upserts return True if a new entry was created, False if an existing
             one was updated.
         """
@@ -637,6 +648,18 @@ class SQLBaseStore(object):
     def _simple_upsert_txn_emulated(
         self, txn, table, keyvalues, values, insertion_values={}, lock=True
     ):
+        """
+        Args:
+            table (str): The table to upsert into
+            keyvalues (dict): The unique key tables and their new values
+            values (dict): The nonunique columns and their new values
+            insertion_values (dict): additional key/values to use only when
+                inserting
+            lock (bool): True to lock the table when doing the upsert.
+        Returns:
+            bool: Return True if a new entry was created, False if an existing
+            one was updated.
+        """
         # We need to lock the table :(, unless we're *really* careful
         if lock:
             self.database_engine.lock_table(txn, table)
-- 
cgit 1.4.1


From 03c85335d1d386c0523af3b6bf992f83bfb905d7 Mon Sep 17 00:00:00 2001
From: Richard van der Hoff <1389908+richvdh@users.noreply.github.com>
Date: Thu, 24 Jan 2019 17:22:09 +0000
Subject: Apply suggestions from code review

Co-Authored-By: anoadragon453 <1342360+anoadragon453@users.noreply.github.com>
---
 synapse/handlers/search.py | 2 +-
 synapse/storage/state.py   | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

(limited to 'synapse/storage')

diff --git a/synapse/handlers/search.py b/synapse/handlers/search.py
index 75c26fe065..49c439313e 100644
--- a/synapse/handlers/search.py
+++ b/synapse/handlers/search.py
@@ -52,7 +52,7 @@ class SearchHandler(BaseHandler):
             room_id (str): id of the room to search through.
 
         Returns:
-            Deferred[iterable[str]]: predecessor room ids
+            Deferred[iterable[unicode]]: predecessor room ids
         """
 
         historical_room_ids = []
diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index fceb9744aa..0a0691cd00 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -446,7 +446,7 @@ class StateGroupWorkerStore(EventsWorkerStore, SQLBaseStore):
             room_id (str)
 
         Returns:
-            Deferred[str]: predecessor room id
+            Deferred[unicode|None]: predecessor room id
         """
         state_ids = yield self.get_current_state_ids(room_id)
         create_id = state_ids.get((EventTypes.Create, ""))
-- 
cgit 1.4.1


From e1781b043b4a73e1b22142b6e09c07ddd782ae57 Mon Sep 17 00:00:00 2001
From: Andrew Morgan <andrew@amorgan.xyz>
Date: Thu, 24 Jan 2019 17:23:39 +0000
Subject: Remove redundant create event None check

---
 synapse/storage/state.py | 3 ---
 1 file changed, 3 deletions(-)

(limited to 'synapse/storage')

diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index fceb9744aa..c02130d9d6 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -458,9 +458,6 @@ class StateGroupWorkerStore(EventsWorkerStore, SQLBaseStore):
         # Retrieve the room's create event
         create_event = yield self.get_event(create_id)
 
-        if not create_event:
-            defer.returnValue(None)
-
         # Return predecessor if present
         defer.returnValue(create_event.content.get("predecessor", None))
 
-- 
cgit 1.4.1


From b8082a54451bb4db30e3b2a4d19dc8cb23330eb7 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 24 Jan 2019 17:33:19 +0000
Subject: Use term 'out of band membership' instead

---
 synapse/events/__init__.py     | 9 +++++----
 synapse/handlers/federation.py | 4 ++--
 synapse/storage/roommember.py  | 6 +++---
 3 files changed, 10 insertions(+), 9 deletions(-)

(limited to 'synapse/storage')

diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py
index 5030636c7e..48289cad06 100644
--- a/synapse/events/__init__.py
+++ b/synapse/events/__init__.py
@@ -41,12 +41,13 @@ class _EventInternalMetadata(object):
     def is_outlier(self):
         return getattr(self, "outlier", False)
 
-    def is_new_remote_event(self):
-        """Whether this is a new remote event, like an invite or an invite
+    def is_out_of_band_membership(self):
+        """Whether this is an out of band membership, like an invite or an invite
         rejection. This is needed as those events are marked as outliers, but
-        they still need to be processed.
+        they still need to be processed as if they're new events (e.g. updating
+        invite state in the database, relaying to clients, etc).
         """
-        return getattr(self, "new_remote_event", False)
+        return getattr(self, "out_of_band_membership", False)
 
     def get_send_on_behalf_of(self):
         """Whether this server should send the event on behalf of another server.
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index e017cab777..242719b7ce 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -1284,7 +1284,7 @@ class FederationHandler(BaseHandler):
             )
 
         event.internal_metadata.outlier = True
-        event.internal_metadata.new_remote_event = True
+        event.internal_metadata.out_of_band_membership = True
 
         event.signatures.update(
             compute_event_signature(
@@ -1310,7 +1310,7 @@ class FederationHandler(BaseHandler):
         # Mark as outlier as we don't have any state for this event; we're not
         # even in the room.
         event.internal_metadata.outlier = True
-        event.internal_metadata.new_remote_event = True
+        event.internal_metadata.out_of_band_membership = True
 
         # Try the host that we succesfully called /make_leave/ on first for
         # the /send_leave/ request.
diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py
index 40b13de80b..592c1bcd33 100644
--- a/synapse/storage/roommember.py
+++ b/synapse/storage/roommember.py
@@ -589,11 +589,11 @@ class RoomMemberStore(RoomMemberWorkerStore):
 
             # We update the local_invites table only if the event is "current",
             # i.e., its something that has just happened. If the event is an
-            # outlier it is only current if its a "new remote event", like a
-            # remote invite or a rejection of a remote invite.
+            # outlier it is only current if its an "out of band membership",
+            # like a remote invite or a rejection of a remote invite.
             is_new_state = not backfilled and (
                 not event.internal_metadata.is_outlier()
-                or event.internal_metadata.is_new_remote_event()
+                or event.internal_metadata.is_out_of_band_membership()
             )
             is_mine = self.hs.is_mine_id(event.state_key)
             if is_new_state and is_mine:
-- 
cgit 1.4.1


From be6a7e47fa75d72466a356c254376b8eb0707bb2 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erikj@jki.re>
Date: Fri, 25 Jan 2019 10:23:51 +0000
Subject: Revert "Require event format version to parse or create events"

---
 changelog.d/4447.misc                   |  1 -
 changelog.d/4451.misc                   |  1 -
 synapse/events/__init__.py              | 39 +----------------
 synapse/events/builder.py               | 51 +----------------------
 synapse/federation/federation_base.py   |  9 ++--
 synapse/federation/federation_client.py | 74 ++++++++++++---------------------
 synapse/federation/federation_server.py | 41 ++++++------------
 synapse/federation/transport/server.py  |  4 +-
 synapse/handlers/federation.py          | 72 +++++++++++++-------------------
 synapse/handlers/message.py             | 10 +----
 synapse/replication/http/federation.py  |  8 +---
 synapse/replication/http/send_event.py  |  8 +---
 synapse/storage/events_worker.py        |  8 +++-
 tests/storage/test_redaction.py         |  5 +--
 tests/storage/test_roommember.py        |  3 +-
 tests/storage/test_state.py             |  3 +-
 tests/test_visibility.py                |  4 --
 tests/utils.py                          |  3 +-
 18 files changed, 91 insertions(+), 253 deletions(-)
 delete mode 100644 changelog.d/4447.misc
 delete mode 100644 changelog.d/4451.misc

(limited to 'synapse/storage')

diff --git a/changelog.d/4447.misc b/changelog.d/4447.misc
deleted file mode 100644
index 43f8963614..0000000000
--- a/changelog.d/4447.misc
+++ /dev/null
@@ -1 +0,0 @@
-Add infrastructure to support different event formats
diff --git a/changelog.d/4451.misc b/changelog.d/4451.misc
deleted file mode 100644
index 43f8963614..0000000000
--- a/changelog.d/4451.misc
+++ /dev/null
@@ -1 +0,0 @@
-Add infrastructure to support different event formats
diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py
index c3e6caf597..888296933b 100644
--- a/synapse/events/__init__.py
+++ b/synapse/events/__init__.py
@@ -18,11 +18,7 @@ from distutils.util import strtobool
 
 import six
 
-from synapse.api.constants import (
-    KNOWN_EVENT_FORMAT_VERSIONS,
-    KNOWN_ROOM_VERSIONS,
-    EventFormatVersions,
-)
+from synapse.api.constants import EventFormatVersions
 from synapse.util.caches import intern_dict
 from synapse.util.frozenutils import freeze
 
@@ -244,36 +240,3 @@ class FrozenEvent(EventBase):
             self.get("type", None),
             self.get("state_key", None),
         )
-
-
-def room_version_to_event_format(room_version):
-    """Converts a room version string to the event format
-
-    Args:
-        room_version (str)
-
-    Returns:
-        int
-    """
-    if room_version not in KNOWN_ROOM_VERSIONS:
-        raise
-
-    return EventFormatVersions.V1
-
-
-def event_type_from_format_version(format_version):
-    """Returns the python type to use to construct an Event object for the
-    given event format version.
-
-    Args:
-        format_version (int): The event format version
-
-    Returns:
-        type: A type that can be initialized as per the initializer of
-        `FrozenEvent`
-    """
-    if format_version not in KNOWN_EVENT_FORMAT_VERSIONS:
-        raise Exception(
-            "No event format %r" % (format_version,)
-        )
-    return FrozenEvent
diff --git a/synapse/events/builder.py b/synapse/events/builder.py
index 7e63371095..e662eaef10 100644
--- a/synapse/events/builder.py
+++ b/synapse/events/builder.py
@@ -15,39 +15,12 @@
 
 import copy
 
-from synapse.api.constants import RoomVersions
 from synapse.types import EventID
 from synapse.util.stringutils import random_string
 
 from . import EventBase, FrozenEvent, _event_dict_property
 
 
-def get_event_builder(room_version, key_values={}, internal_metadata_dict={}):
-    """Generate an event builder appropriate for the given room version
-
-    Args:
-        room_version (str): Version of the room that we're creating an
-            event builder for
-        key_values (dict): Fields used as the basis of the new event
-        internal_metadata_dict (dict): Used to create the `_EventInternalMetadata`
-            object.
-
-    Returns:
-        EventBuilder
-    """
-    if room_version in {
-        RoomVersions.V1,
-        RoomVersions.V2,
-        RoomVersions.VDH_TEST,
-        RoomVersions.STATE_V2_TEST,
-    }:
-        return EventBuilder(key_values, internal_metadata_dict)
-    else:
-        raise Exception(
-            "No event format defined for version %r" % (room_version,)
-        )
-
-
 class EventBuilder(EventBase):
     def __init__(self, key_values={}, internal_metadata_dict={}):
         signatures = copy.deepcopy(key_values.pop("signatures", {}))
@@ -85,29 +58,7 @@ class EventBuilderFactory(object):
 
         return e_id.to_string()
 
-    def new(self, room_version, key_values={}):
-        """Generate an event builder appropriate for the given room version
-
-        Args:
-            room_version (str): Version of the room that we're creating an
-                event builder for
-            key_values (dict): Fields used as the basis of the new event
-
-        Returns:
-            EventBuilder
-        """
-
-        # There's currently only the one event version defined
-        if room_version not in {
-            RoomVersions.V1,
-            RoomVersions.V2,
-            RoomVersions.VDH_TEST,
-            RoomVersions.STATE_V2_TEST,
-        }:
-            raise Exception(
-                "No event format defined for version %r" % (room_version,)
-            )
-
+    def new(self, key_values={}):
         key_values["event_id"] = self.create_event_id()
 
         time_now = int(self.clock.time_msec())
diff --git a/synapse/federation/federation_base.py b/synapse/federation/federation_base.py
index 5c31e5f85f..d749bfdd3a 100644
--- a/synapse/federation/federation_base.py
+++ b/synapse/federation/federation_base.py
@@ -23,7 +23,7 @@ from twisted.internet.defer import DeferredList
 from synapse.api.constants import MAX_DEPTH, EventTypes, Membership
 from synapse.api.errors import Codes, SynapseError
 from synapse.crypto.event_signing import check_event_content_hash
-from synapse.events import event_type_from_format_version
+from synapse.events import FrozenEvent
 from synapse.events.utils import prune_event
 from synapse.http.servlet import assert_params_in_dict
 from synapse.types import get_domain_from_id
@@ -302,12 +302,11 @@ def _is_invite_via_3pid(event):
     )
 
 
-def event_from_pdu_json(pdu_json, event_format_version, outlier=False):
+def event_from_pdu_json(pdu_json, outlier=False):
     """Construct a FrozenEvent from an event json received over federation
 
     Args:
         pdu_json (object): pdu as received over federation
-        event_format_version (int): The event format version
         outlier (bool): True to mark this event as an outlier
 
     Returns:
@@ -331,8 +330,8 @@ def event_from_pdu_json(pdu_json, event_format_version, outlier=False):
     elif depth > MAX_DEPTH:
         raise SynapseError(400, "Depth too large", Codes.BAD_JSON)
 
-    event = event_type_from_format_version(event_format_version)(
-        pdu_json,
+    event = FrozenEvent(
+        pdu_json
     )
 
     event.internal_metadata.outlier = outlier
diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py
index 4b25f891ca..777deabdf7 100644
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -38,7 +38,6 @@ from synapse.api.errors import (
     SynapseError,
 )
 from synapse.crypto.event_signing import add_hashes_and_signatures
-from synapse.events import room_version_to_event_format
 from synapse.federation.federation_base import FederationBase, event_from_pdu_json
 from synapse.util import logcontext, unwrapFirstError
 from synapse.util.caches.expiringcache import ExpiringCache
@@ -170,13 +169,13 @@ class FederationClient(FederationBase):
 
     @defer.inlineCallbacks
     @log_function
-    def backfill(self, dest, room_id, limit, extremities):
+    def backfill(self, dest, context, limit, extremities):
         """Requests some more historic PDUs for the given context from the
         given destination server.
 
         Args:
             dest (str): The remote home server to ask.
-            room_id (str): The room_id to backfill.
+            context (str): The context to backfill.
             limit (int): The maximum number of PDUs to return.
             extremities (list): List of PDU id and origins of the first pdus
                 we have seen from the context
@@ -191,15 +190,12 @@ class FederationClient(FederationBase):
             return
 
         transaction_data = yield self.transport_layer.backfill(
-            dest, room_id, extremities, limit)
+            dest, context, extremities, limit)
 
         logger.debug("backfill transaction_data=%s", repr(transaction_data))
 
-        room_version = yield self.store.get_room_version(room_id)
-        format_ver = room_version_to_event_format(room_version)
-
         pdus = [
-            event_from_pdu_json(p, format_ver, outlier=False)
+            event_from_pdu_json(p, outlier=False)
             for p in transaction_data["pdus"]
         ]
 
@@ -243,8 +239,6 @@ class FederationClient(FederationBase):
 
         pdu_attempts = self.pdu_destination_tried.setdefault(event_id, {})
 
-        format_ver = room_version_to_event_format(room_version)
-
         signed_pdu = None
         for destination in destinations:
             now = self._clock.time_msec()
@@ -260,7 +254,7 @@ class FederationClient(FederationBase):
                 logger.debug("transaction_data %r", transaction_data)
 
                 pdu_list = [
-                    event_from_pdu_json(p, format_ver, outlier=outlier)
+                    event_from_pdu_json(p, outlier=outlier)
                     for p in transaction_data["pdus"]
                 ]
 
@@ -354,16 +348,12 @@ class FederationClient(FederationBase):
             destination, room_id, event_id=event_id,
         )
 
-        room_version = yield self.store.get_room_version(room_id)
-        format_ver = room_version_to_event_format(room_version)
-
         pdus = [
-            event_from_pdu_json(p, format_ver, outlier=True)
-            for p in result["pdus"]
+            event_from_pdu_json(p, outlier=True) for p in result["pdus"]
         ]
 
         auth_chain = [
-            event_from_pdu_json(p, format_ver, outlier=True)
+            event_from_pdu_json(p, outlier=True)
             for p in result.get("auth_chain", [])
         ]
 
@@ -371,6 +361,8 @@ class FederationClient(FederationBase):
             ev.event_id for ev in itertools.chain(pdus, auth_chain)
         ])
 
+        room_version = yield self.store.get_room_version(room_id)
+
         signed_pdus = yield self._check_sigs_and_hash_and_fetch(
             destination,
             [p for p in pdus if p.event_id not in seen_events],
@@ -469,14 +461,13 @@ class FederationClient(FederationBase):
             destination, room_id, event_id,
         )
 
-        room_version = yield self.store.get_room_version(room_id)
-        format_ver = room_version_to_event_format(room_version)
-
         auth_chain = [
-            event_from_pdu_json(p, format_ver, outlier=True)
+            event_from_pdu_json(p, outlier=True)
             for p in res["auth_chain"]
         ]
 
+        room_version = yield self.store.get_room_version(room_id)
+
         signed_auth = yield self._check_sigs_and_hash_and_fetch(
             destination, auth_chain,
             outlier=True, room_version=room_version,
@@ -566,9 +557,9 @@ class FederationClient(FederationBase):
             params (dict[str, str|Iterable[str]]): Query parameters to include in the
                 request.
         Return:
-            Deferred[tuple[str, FrozenEvent, int]]: resolves to a tuple of
-            `(origin, event, event_format)` where origin is the remote
-            homeserver which generated the event.
+            Deferred[tuple[str, FrozenEvent]]: resolves to a tuple of `origin`
+            and event where origin is the remote homeserver which generated
+            the event.
 
             Fails with a ``SynapseError`` if the chosen remote server
             returns a 300/400 code.
@@ -588,11 +579,6 @@ class FederationClient(FederationBase):
                 destination, room_id, user_id, membership, params,
             )
 
-            # Note: If not supplied, the room version may be either v1 or v2,
-            # however either way the event format version will be v1.
-            room_version = ret.get("room_version", RoomVersions.V1)
-            event_format = room_version_to_event_format(room_version)
-
             pdu_dict = ret.get("event", None)
             if not isinstance(pdu_dict, dict):
                 raise InvalidResponseError("Bad 'event' field in response")
@@ -612,7 +598,7 @@ class FederationClient(FederationBase):
             pdu_dict.pop("origin_server_ts", None)
             pdu_dict.pop("unsigned", None)
 
-            builder = self.event_builder_factory.new(room_version, pdu_dict)
+            builder = self.event_builder_factory.new(pdu_dict)
             add_hashes_and_signatures(
                 builder,
                 self.hs.hostname,
@@ -621,14 +607,14 @@ class FederationClient(FederationBase):
             ev = builder.build()
 
             defer.returnValue(
-                (destination, ev, event_format)
+                (destination, ev)
             )
 
         return self._try_destination_list(
             "make_" + membership, destinations, send_request,
         )
 
-    def send_join(self, destinations, pdu, event_format_version):
+    def send_join(self, destinations, pdu):
         """Sends a join event to one of a list of homeservers.
 
         Doing so will cause the remote server to add the event to the graph,
@@ -638,7 +624,6 @@ class FederationClient(FederationBase):
             destinations (str): Candidate homeservers which are probably
                 participating in the room.
             pdu (BaseEvent): event to be sent
-            event_format_version (int): The event format version
 
         Return:
             Deferred: resolves to a dict with members ``origin`` (a string
@@ -684,12 +669,12 @@ class FederationClient(FederationBase):
             logger.debug("Got content: %s", content)
 
             state = [
-                event_from_pdu_json(p, event_format_version, outlier=True)
+                event_from_pdu_json(p, outlier=True)
                 for p in content.get("state", [])
             ]
 
             auth_chain = [
-                event_from_pdu_json(p, event_format_version, outlier=True)
+                event_from_pdu_json(p, outlier=True)
                 for p in content.get("auth_chain", [])
             ]
 
@@ -767,10 +752,7 @@ class FederationClient(FederationBase):
 
         logger.debug("Got response to send_invite: %s", pdu_dict)
 
-        room_version = yield self.store.get_room_version(room_id)
-        format_ver = room_version_to_event_format(room_version)
-
-        pdu = event_from_pdu_json(pdu_dict, format_ver)
+        pdu = event_from_pdu_json(pdu_dict)
 
         # Check signatures are correct.
         pdu = yield self._check_sigs_and_hash(pdu)
@@ -848,14 +830,13 @@ class FederationClient(FederationBase):
             content=send_content,
         )
 
-        room_version = yield self.store.get_room_version(room_id)
-        format_ver = room_version_to_event_format(room_version)
-
         auth_chain = [
-            event_from_pdu_json(e, format_ver)
+            event_from_pdu_json(e)
             for e in content["auth_chain"]
         ]
 
+        room_version = yield self.store.get_room_version(room_id)
+
         signed_auth = yield self._check_sigs_and_hash_and_fetch(
             destination, auth_chain, outlier=True, room_version=room_version,
         )
@@ -899,14 +880,13 @@ class FederationClient(FederationBase):
                 timeout=timeout,
             )
 
-            room_version = yield self.store.get_room_version(room_id)
-            format_ver = room_version_to_event_format(room_version)
-
             events = [
-                event_from_pdu_json(e, format_ver)
+                event_from_pdu_json(e)
                 for e in content.get("events", [])
             ]
 
+            room_version = yield self.store.get_room_version(room_id)
+
             signed_events = yield self._check_sigs_and_hash_and_fetch(
                 destination, events, outlier=False, room_version=room_version,
             )
diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index 4aa04b9588..cb729c69ea 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -34,7 +34,6 @@ from synapse.api.errors import (
     SynapseError,
 )
 from synapse.crypto.event_signing import compute_event_signature
-from synapse.events import room_version_to_event_format
 from synapse.federation.federation_base import FederationBase, event_from_pdu_json
 from synapse.federation.persistence import TransactionActions
 from synapse.federation.units import Edu, Transaction
@@ -179,13 +178,14 @@ class FederationServer(FederationBase):
                 continue
 
             try:
-                room_version = yield self.store.get_room_version(room_id)
-                format_ver = room_version_to_event_format(room_version)
+                # In future we will actually use the room version to parse the
+                # PDU into an event.
+                yield self.store.get_room_version(room_id)
             except NotFoundError:
                 logger.info("Ignoring PDU for unknown room_id: %s", room_id)
                 continue
 
-            event = event_from_pdu_json(p, format_ver)
+            event = event_from_pdu_json(p)
             pdus_by_room.setdefault(room_id, []).append(event)
 
         pdu_results = {}
@@ -370,9 +370,7 @@ class FederationServer(FederationBase):
 
     @defer.inlineCallbacks
     def on_invite_request(self, origin, content, room_version):
-        format_ver = room_version_to_event_format(room_version)
-
-        pdu = event_from_pdu_json(content, format_ver)
+        pdu = event_from_pdu_json(content)
         origin_host, _ = parse_server_name(origin)
         yield self.check_server_matches_acl(origin_host, pdu.room_id)
         ret_pdu = yield self.handler.on_invite_request(origin, pdu)
@@ -380,12 +378,9 @@ class FederationServer(FederationBase):
         defer.returnValue({"event": ret_pdu.get_pdu_json(time_now)})
 
     @defer.inlineCallbacks
-    def on_send_join_request(self, origin, content, room_id):
+    def on_send_join_request(self, origin, content):
         logger.debug("on_send_join_request: content: %s", content)
-
-        room_version = yield self.store.get_room_version(room_id)
-        format_ver = room_version_to_event_format(room_version)
-        pdu = event_from_pdu_json(content, format_ver)
+        pdu = event_from_pdu_json(content)
 
         origin_host, _ = parse_server_name(origin)
         yield self.check_server_matches_acl(origin_host, pdu.room_id)
@@ -405,22 +400,13 @@ class FederationServer(FederationBase):
         origin_host, _ = parse_server_name(origin)
         yield self.check_server_matches_acl(origin_host, room_id)
         pdu = yield self.handler.on_make_leave_request(room_id, user_id)
-
-        room_version = yield self.store.get_room_version(room_id)
-
         time_now = self._clock.time_msec()
-        defer.returnValue({
-            "event": pdu.get_pdu_json(time_now),
-            "room_version": room_version,
-        })
+        defer.returnValue({"event": pdu.get_pdu_json(time_now)})
 
     @defer.inlineCallbacks
-    def on_send_leave_request(self, origin, content, room_id):
+    def on_send_leave_request(self, origin, content):
         logger.debug("on_send_leave_request: content: %s", content)
-
-        room_version = yield self.store.get_room_version(room_id)
-        format_ver = room_version_to_event_format(room_version)
-        pdu = event_from_pdu_json(content, format_ver)
+        pdu = event_from_pdu_json(content)
 
         origin_host, _ = parse_server_name(origin)
         yield self.check_server_matches_acl(origin_host, pdu.room_id)
@@ -466,14 +452,13 @@ class FederationServer(FederationBase):
             origin_host, _ = parse_server_name(origin)
             yield self.check_server_matches_acl(origin_host, room_id)
 
-            room_version = yield self.store.get_room_version(room_id)
-            format_ver = room_version_to_event_format(room_version)
-
             auth_chain = [
-                event_from_pdu_json(e, format_ver)
+                event_from_pdu_json(e)
                 for e in content["auth_chain"]
             ]
 
+            room_version = yield self.store.get_room_version(room_id)
+
             signed_auth = yield self._check_sigs_and_hash_and_fetch(
                 origin, auth_chain, outlier=True, room_version=room_version,
             )
diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py
index 67ae0212c3..4557a9e66e 100644
--- a/synapse/federation/transport/server.py
+++ b/synapse/federation/transport/server.py
@@ -469,7 +469,7 @@ class FederationSendLeaveServlet(BaseFederationServlet):
 
     @defer.inlineCallbacks
     def on_PUT(self, origin, content, query, room_id, event_id):
-        content = yield self.handler.on_send_leave_request(origin, content, room_id)
+        content = yield self.handler.on_send_leave_request(origin, content)
         defer.returnValue((200, content))
 
 
@@ -487,7 +487,7 @@ class FederationSendJoinServlet(BaseFederationServlet):
     def on_PUT(self, origin, content, query, context, event_id):
         # TODO(paul): assert that context/event_id parsed from path actually
         #   match those given in content
-        content = yield self.handler.on_send_join_request(origin, content, context)
+        content = yield self.handler.on_send_join_request(origin, content)
         defer.returnValue((200, content))
 
 
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index a4b771049c..453d393ce1 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -1061,7 +1061,7 @@ class FederationHandler(BaseHandler):
         """
         logger.debug("Joining %s to %s", joinee, room_id)
 
-        origin, event, event_format_version = yield self._make_and_verify_event(
+        origin, event = yield self._make_and_verify_event(
             target_hosts,
             room_id,
             joinee,
@@ -1091,9 +1091,7 @@ class FederationHandler(BaseHandler):
                 target_hosts.insert(0, origin)
             except ValueError:
                 pass
-            ret = yield self.federation_client.send_join(
-                target_hosts, event, event_format_version,
-            )
+            ret = yield self.federation_client.send_join(target_hosts, event)
 
             origin = ret["origin"]
             state = ret["state"]
@@ -1166,18 +1164,13 @@ class FederationHandler(BaseHandler):
         """
         event_content = {"membership": Membership.JOIN}
 
-        room_version = yield self.store.get_room_version(room_id)
-
-        builder = self.event_builder_factory.new(
-            room_version,
-            {
-                "type": EventTypes.Member,
-                "content": event_content,
-                "room_id": room_id,
-                "sender": user_id,
-                "state_key": user_id,
-            }
-        )
+        builder = self.event_builder_factory.new({
+            "type": EventTypes.Member,
+            "content": event_content,
+            "room_id": room_id,
+            "sender": user_id,
+            "state_key": user_id,
+        })
 
         try:
             event, context = yield self.event_creation_handler.create_new_client_event(
@@ -1311,7 +1304,7 @@ class FederationHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def do_remotely_reject_invite(self, target_hosts, room_id, user_id):
-        origin, event, event_format_version = yield self._make_and_verify_event(
+        origin, event = yield self._make_and_verify_event(
             target_hosts,
             room_id,
             user_id,
@@ -1343,7 +1336,7 @@ class FederationHandler(BaseHandler):
     @defer.inlineCallbacks
     def _make_and_verify_event(self, target_hosts, room_id, user_id, membership,
                                content={}, params=None):
-        origin, event, format_ver = yield self.federation_client.make_membership_event(
+        origin, pdu = yield self.federation_client.make_membership_event(
             target_hosts,
             room_id,
             user_id,
@@ -1352,7 +1345,9 @@ class FederationHandler(BaseHandler):
             params=params,
         )
 
-        logger.debug("Got response to make_%s: %s", membership, event)
+        logger.debug("Got response to make_%s: %s", membership, pdu)
+
+        event = pdu
 
         # We should assert some things.
         # FIXME: Do this in a nicer way
@@ -1360,7 +1355,7 @@ class FederationHandler(BaseHandler):
         assert(event.user_id == user_id)
         assert(event.state_key == user_id)
         assert(event.room_id == room_id)
-        defer.returnValue((origin, event, format_ver))
+        defer.returnValue((origin, event))
 
     @defer.inlineCallbacks
     @log_function
@@ -1369,17 +1364,13 @@ class FederationHandler(BaseHandler):
         leave event for the room and return that. We do *not* persist or
         process it until the other server has signed it and sent it back.
         """
-        room_version = yield self.store.get_room_version(room_id)
-        builder = self.event_builder_factory.new(
-            room_version,
-            {
-                "type": EventTypes.Member,
-                "content": {"membership": Membership.LEAVE},
-                "room_id": room_id,
-                "sender": user_id,
-                "state_key": user_id,
-            }
-        )
+        builder = self.event_builder_factory.new({
+            "type": EventTypes.Member,
+            "content": {"membership": Membership.LEAVE},
+            "room_id": room_id,
+            "sender": user_id,
+            "state_key": user_id,
+        })
 
         event, context = yield self.event_creation_handler.create_new_client_event(
             builder=builder,
@@ -2275,16 +2266,14 @@ class FederationHandler(BaseHandler):
         }
 
         if (yield self.auth.check_host_in_room(room_id, self.hs.hostname)):
-            room_version = yield self.store.get_room_version(room_id)
-            builder = self.event_builder_factory.new(room_version, event_dict)
-
+            builder = self.event_builder_factory.new(event_dict)
             EventValidator().validate_new(builder)
             event, context = yield self.event_creation_handler.create_new_client_event(
                 builder=builder
             )
 
             event, context = yield self.add_display_name_to_third_party_invite(
-                room_version, event_dict, event, context
+                event_dict, event, context
             )
 
             try:
@@ -2315,18 +2304,14 @@ class FederationHandler(BaseHandler):
         Returns:
             Deferred: resolves (to None)
         """
-        room_version = yield self.store.get_room_version(room_id)
-
-        # NB: event_dict has a particular specced format we might need to fudge
-        # if we change event formats too much.
-        builder = self.event_builder_factory.new(room_version, event_dict)
+        builder = self.event_builder_factory.new(event_dict)
 
         event, context = yield self.event_creation_handler.create_new_client_event(
             builder=builder,
         )
 
         event, context = yield self.add_display_name_to_third_party_invite(
-            room_version, event_dict, event, context
+            event_dict, event, context
         )
 
         try:
@@ -2346,8 +2331,7 @@ class FederationHandler(BaseHandler):
         yield member_handler.send_membership_event(None, event, context)
 
     @defer.inlineCallbacks
-    def add_display_name_to_third_party_invite(self, room_version, event_dict,
-                                               event, context):
+    def add_display_name_to_third_party_invite(self, event_dict, event, context):
         key = (
             EventTypes.ThirdPartyInvite,
             event.content["third_party_invite"]["signed"]["token"]
@@ -2371,7 +2355,7 @@ class FederationHandler(BaseHandler):
             # auth checks. If we need the invite and don't have it then the
             # auth check code will explode appropriately.
 
-        builder = self.event_builder_factory.new(room_version, event_dict)
+        builder = self.event_builder_factory.new(event_dict)
         EventValidator().validate_new(builder)
         event, context = yield self.event_creation_handler.create_new_client_event(
             builder=builder,
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 7aaa4fba33..a7cd779b02 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -278,15 +278,7 @@ class EventCreationHandler(object):
         """
         yield self.auth.check_auth_blocking(requester.user.to_string())
 
-        if event_dict["type"] == EventTypes.Create and event_dict["state_key"] == "":
-            room_version = event_dict["content"]["room_version"]
-        else:
-            try:
-                room_version = yield self.store.get_room_version(event_dict["room_id"])
-            except NotFoundError:
-                raise AuthError(403, "Unknown room")
-
-        builder = self.event_builder_factory.new(room_version, event_dict)
+        builder = self.event_builder_factory.new(event_dict)
 
         self.validator.validate_new(builder)
 
diff --git a/synapse/replication/http/federation.py b/synapse/replication/http/federation.py
index 2e16c69666..64a79da162 100644
--- a/synapse/replication/http/federation.py
+++ b/synapse/replication/http/federation.py
@@ -17,7 +17,7 @@ import logging
 
 from twisted.internet import defer
 
-from synapse.events import event_type_from_format_version
+from synapse.events import FrozenEvent
 from synapse.events.snapshot import EventContext
 from synapse.http.servlet import parse_json_object_from_request
 from synapse.replication.http._base import ReplicationEndpoint
@@ -70,7 +70,6 @@ class ReplicationFederationSendEventsRestServlet(ReplicationEndpoint):
 
             event_payloads.append({
                 "event": event.get_pdu_json(),
-                "event_format_version": event.format_version,
                 "internal_metadata": event.internal_metadata.get_dict(),
                 "rejected_reason": event.rejected_reason,
                 "context": serialized_context,
@@ -95,12 +94,9 @@ class ReplicationFederationSendEventsRestServlet(ReplicationEndpoint):
             event_and_contexts = []
             for event_payload in event_payloads:
                 event_dict = event_payload["event"]
-                format_ver = content["event_format_version"]
                 internal_metadata = event_payload["internal_metadata"]
                 rejected_reason = event_payload["rejected_reason"]
-
-                EventType = event_type_from_format_version(format_ver)
-                event = EventType(event_dict, internal_metadata, rejected_reason)
+                event = FrozenEvent(event_dict, internal_metadata, rejected_reason)
 
                 context = yield EventContext.deserialize(
                     self.store, event_payload["context"],
diff --git a/synapse/replication/http/send_event.py b/synapse/replication/http/send_event.py
index 3635015eda..5b52c91650 100644
--- a/synapse/replication/http/send_event.py
+++ b/synapse/replication/http/send_event.py
@@ -17,7 +17,7 @@ import logging
 
 from twisted.internet import defer
 
-from synapse.events import event_type_from_format_version
+from synapse.events import FrozenEvent
 from synapse.events.snapshot import EventContext
 from synapse.http.servlet import parse_json_object_from_request
 from synapse.replication.http._base import ReplicationEndpoint
@@ -74,7 +74,6 @@ class ReplicationSendEventRestServlet(ReplicationEndpoint):
 
         payload = {
             "event": event.get_pdu_json(),
-            "event_format_version": event.format_version,
             "internal_metadata": event.internal_metadata.get_dict(),
             "rejected_reason": event.rejected_reason,
             "context": serialized_context,
@@ -91,12 +90,9 @@ class ReplicationSendEventRestServlet(ReplicationEndpoint):
             content = parse_json_object_from_request(request)
 
             event_dict = content["event"]
-            format_ver = content["event_format_version"]
             internal_metadata = content["internal_metadata"]
             rejected_reason = content["rejected_reason"]
-
-            EventType = event_type_from_format_version(format_ver)
-            event = EventType(event_dict, internal_metadata, rejected_reason)
+            event = FrozenEvent(event_dict, internal_metadata, rejected_reason)
 
             requester = Requester.deserialize(self.store, content["requester"])
             context = yield EventContext.deserialize(self.store, content["context"])
diff --git a/synapse/storage/events_worker.py b/synapse/storage/events_worker.py
index 0a0ca58fc4..599f892858 100644
--- a/synapse/storage/events_worker.py
+++ b/synapse/storage/events_worker.py
@@ -23,7 +23,7 @@ from twisted.internet import defer
 
 from synapse.api.constants import EventFormatVersions
 from synapse.api.errors import NotFoundError
-from synapse.events import FrozenEvent, event_type_from_format_version  # noqa: F401
+from synapse.events import FrozenEvent
 # these are only included to make the type annotations work
 from synapse.events.snapshot import EventContext  # noqa: F401
 from synapse.events.utils import prune_event
@@ -412,7 +412,11 @@ class EventsWorkerStore(SQLBaseStore):
                 # of a event format version, so it must be a V1 event.
                 format_version = EventFormatVersions.V1
 
-            original_ev = event_type_from_format_version(format_version)(
+            # TODO: When we implement new event formats we'll need to use a
+            # different event python type
+            assert format_version == EventFormatVersions.V1
+
+            original_ev = FrozenEvent(
                 event_dict=d,
                 internal_metadata_dict=internal_metadata,
                 rejected_reason=rejected_reason,
diff --git a/tests/storage/test_redaction.py b/tests/storage/test_redaction.py
index 3957561b1e..02bf975fbf 100644
--- a/tests/storage/test_redaction.py
+++ b/tests/storage/test_redaction.py
@@ -18,7 +18,7 @@ from mock import Mock
 
 from twisted.internet import defer
 
-from synapse.api.constants import EventTypes, Membership, RoomVersions
+from synapse.api.constants import EventTypes, Membership
 from synapse.types import RoomID, UserID
 
 from tests import unittest
@@ -52,7 +52,6 @@ class RedactionTestCase(unittest.TestCase):
         content = {"membership": membership}
         content.update(extra_content)
         builder = self.event_builder_factory.new(
-            RoomVersions.V1,
             {
                 "type": EventTypes.Member,
                 "sender": user.to_string(),
@@ -75,7 +74,6 @@ class RedactionTestCase(unittest.TestCase):
         self.depth += 1
 
         builder = self.event_builder_factory.new(
-            RoomVersions.V1,
             {
                 "type": EventTypes.Message,
                 "sender": user.to_string(),
@@ -96,7 +94,6 @@ class RedactionTestCase(unittest.TestCase):
     @defer.inlineCallbacks
     def inject_redaction(self, room, event_id, user, reason):
         builder = self.event_builder_factory.new(
-            RoomVersions.V1,
             {
                 "type": EventTypes.Redaction,
                 "sender": user.to_string(),
diff --git a/tests/storage/test_roommember.py b/tests/storage/test_roommember.py
index 7fa2f4fd70..978c66133d 100644
--- a/tests/storage/test_roommember.py
+++ b/tests/storage/test_roommember.py
@@ -18,7 +18,7 @@ from mock import Mock
 
 from twisted.internet import defer
 
-from synapse.api.constants import EventTypes, Membership, RoomVersions
+from synapse.api.constants import EventTypes, Membership
 from synapse.types import RoomID, UserID
 
 from tests import unittest
@@ -50,7 +50,6 @@ class RoomMemberStoreTestCase(unittest.TestCase):
     @defer.inlineCallbacks
     def inject_room_member(self, room, user, membership, replaces_state=None):
         builder = self.event_builder_factory.new(
-            RoomVersions.V1,
             {
                 "type": EventTypes.Member,
                 "sender": user.to_string(),
diff --git a/tests/storage/test_state.py b/tests/storage/test_state.py
index a1f99134dc..086a39d834 100644
--- a/tests/storage/test_state.py
+++ b/tests/storage/test_state.py
@@ -17,7 +17,7 @@ import logging
 
 from twisted.internet import defer
 
-from synapse.api.constants import EventTypes, Membership, RoomVersions
+from synapse.api.constants import EventTypes, Membership
 from synapse.storage.state import StateFilter
 from synapse.types import RoomID, UserID
 
@@ -52,7 +52,6 @@ class StateStoreTestCase(tests.unittest.TestCase):
     @defer.inlineCallbacks
     def inject_state_event(self, room, sender, typ, state_key, content):
         builder = self.event_builder_factory.new(
-            RoomVersions.V1,
             {
                 "type": typ,
                 "sender": sender.to_string(),
diff --git a/tests/test_visibility.py b/tests/test_visibility.py
index 82d63ce00e..2eea3b098b 100644
--- a/tests/test_visibility.py
+++ b/tests/test_visibility.py
@@ -17,7 +17,6 @@ import logging
 from twisted.internet import defer
 from twisted.internet.defer import succeed
 
-from synapse.api.constants import RoomVersions
 from synapse.events import FrozenEvent
 from synapse.visibility import filter_events_for_server
 
@@ -125,7 +124,6 @@ class FilterEventsForServerTestCase(tests.unittest.TestCase):
     def inject_visibility(self, user_id, visibility):
         content = {"history_visibility": visibility}
         builder = self.event_builder_factory.new(
-            RoomVersions.V1,
             {
                 "type": "m.room.history_visibility",
                 "sender": user_id,
@@ -146,7 +144,6 @@ class FilterEventsForServerTestCase(tests.unittest.TestCase):
         content = {"membership": membership}
         content.update(extra_content)
         builder = self.event_builder_factory.new(
-            RoomVersions.V1,
             {
                 "type": "m.room.member",
                 "sender": user_id,
@@ -168,7 +165,6 @@ class FilterEventsForServerTestCase(tests.unittest.TestCase):
         if content is None:
             content = {"body": "testytest"}
         builder = self.event_builder_factory.new(
-            RoomVersions.V1,
             {
                 "type": "m.room.message",
                 "sender": user_id,
diff --git a/tests/utils.py b/tests/utils.py
index 2dfcb70a93..df73c539c3 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -26,7 +26,7 @@ from six.moves.urllib import parse as urlparse
 
 from twisted.internet import defer, reactor
 
-from synapse.api.constants import EventTypes, RoomVersions
+from synapse.api.constants import EventTypes
 from synapse.api.errors import CodeMessageException, cs_error
 from synapse.config.server import ServerConfig
 from synapse.federation.transport import server
@@ -624,7 +624,6 @@ def create_room(hs, room_id, creator_id):
     event_creation_handler = hs.get_event_creation_handler()
 
     builder = event_builder_factory.new(
-        RoomVersions.V1,
         {
             "type": EventTypes.Create,
             "state_key": "",
-- 
cgit 1.4.1