Fetch events in a separate transaction.

This has a couple of benefits:

- It reduces the time of transactions, allowing other database requests
  to run.
- Fetching events is given a dedicated database thread, and so can't
  starve other database requests.

synapse/storage/stream.py

@@ -220,8 +220,11 @@ class StreamStore(SQLBaseStore):
 
             rows = self.cursor_to_dict(txn)
 
-            ret = self._get_events_txn(
-                txn,
+            return rows
+
+        rows = yield self.runInteraction("get_room_events_stream_for_room", f)
+
+        ret = yield self._get_events(
             [r["event_id"] for r in rows],
             get_prev_content=True
         )
@@ -237,10 +240,9 @@ class StreamStore(SQLBaseStore):
             # get.
             key = from_key
 
-            return ret, key
-        res = yield self.runInteraction("get_room_events_stream_for_room", f)
-        defer.returnValue(res)
+        defer.returnValue((ret, key))
 
+    @defer.inlineCallbacks
     def get_room_changes_for_user(self, user_id, from_key, to_key):
         if from_key is not None:
             from_id = RoomStreamToken.parse_stream_token(from_key).stream
@@ -249,14 +251,14 @@ class StreamStore(SQLBaseStore):
         to_id = RoomStreamToken.parse_stream_token(to_key).stream
 
         if from_key == to_key:
-            return defer.succeed([])
+            defer.returnValue([])
 
         if from_id:
             has_changed = self._membership_stream_cache.has_entity_changed(
                 user_id, int(from_id)
             )
             if not has_changed:
-                return defer.succeed([])
+                defer.returnValue([])
 
         def f(txn):
             if from_id is not None:
@@ -281,17 +283,18 @@ class StreamStore(SQLBaseStore):
                 txn.execute(sql, (user_id, to_id,))
             rows = self.cursor_to_dict(txn)
 
-            ret = self._get_events_txn(
-                txn,
+            return rows
+
+        rows = yield self.runInteraction("get_room_changes_for_user", f)
+
+        ret = yield self._get_events(
             [r["event_id"] for r in rows],
             get_prev_content=True
         )
 
         self._set_before_and_after(ret, rows, topo_order=False)
 
-            return ret
-
-        return self.runInteraction("get_room_changes_for_user", f)
+        defer.returnValue(ret)
 
     def get_room_events_stream(
         self,