Merge branch 'develop' of github.com:matrix-org/synapse into babolivier/account_expiration

1 files changed, 210 insertions, 275 deletions

diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 7e3903859b..983ce026e1 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -41,7 +41,7 @@ try:
     MAX_TXN_ID = sys.maxint - 1
 except AttributeError:
     # python 3 does not have a maximum int value
-    MAX_TXN_ID = 2**63 - 1
+    MAX_TXN_ID = 2 ** 63 - 1
 
 sql_logger = logging.getLogger("synapse.storage.SQL")
 transaction_logger = logging.getLogger("synapse.storage.txn")
@@ -76,12 +76,18 @@ class LoggingTransaction(object):
     """An object that almost-transparently proxies for the 'txn' object
     passed to the constructor. Adds logging and metrics to the .execute()
     method."""
+
     __slots__ = [
-        "txn", "name", "database_engine", "after_callbacks", "exception_callbacks",
+        "txn",
+        "name",
+        "database_engine",
+        "after_callbacks",
+        "exception_callbacks",
     ]
 
-    def __init__(self, txn, name, database_engine, after_callbacks,
-                 exception_callbacks):
+    def __init__(
+        self, txn, name, database_engine, after_callbacks, exception_callbacks
+    ):
         object.__setattr__(self, "txn", txn)
         object.__setattr__(self, "name", name)
         object.__setattr__(self, "database_engine", database_engine)
@@ -110,6 +116,7 @@ class LoggingTransaction(object):
     def execute_batch(self, sql, args):
         if isinstance(self.database_engine, PostgresEngine):
             from psycopg2.extras import execute_batch
+
             self._do_execute(lambda *x: execute_batch(self.txn, *x), sql, args)
         else:
             for val in args:
@@ -134,10 +141,7 @@ class LoggingTransaction(object):
         sql = self.database_engine.convert_param_style(sql)
         if args:
             try:
-                sql_logger.debug(
-                    "[SQL values] {%s} %r",
-                    self.name, args[0]
-                )
+                sql_logger.debug("[SQL values] {%s} %r", self.name, args[0])
             except Exception:
                 # Don't let logging failures stop SQL from working
                 pass
@@ -145,9 +149,7 @@ class LoggingTransaction(object):
         start = time.time()
 
         try:
-            return func(
-                sql, *args
-            )
+            return func(sql, *args)
         except Exception as e:
             logger.debug("[SQL FAIL] {%s} %s", self.name, e)
             raise
@@ -176,11 +178,9 @@ class PerformanceCounters(object):
         counters = []
         for name, (count, cum_time) in iteritems(self.current_counters):
             prev_count, prev_time = self.previous_counters.get(name, (0, 0))
-            counters.append((
-                (cum_time - prev_time) / interval_duration,
-                count - prev_count,
-                name
-            ))
+            counters.append(
+                ((cum_time - prev_time) / interval_duration, count - prev_count, name)
+            )
 
         self.previous_counters = dict(self.current_counters)
 
@@ -212,8 +212,9 @@ class SQLBaseStore(object):
         self._txn_perf_counters = PerformanceCounters()
         self._get_event_counters = PerformanceCounters()
 
-        self._get_event_cache = Cache("*getEvent*", keylen=3,
-                                      max_entries=hs.config.event_cache_size)
+        self._get_event_cache = Cache(
+            "*getEvent*", keylen=3, max_entries=hs.config.event_cache_size
+        )
 
         self._event_fetch_lock = threading.Condition()
         self._event_fetch_list = []
@@ -239,7 +240,7 @@ class SQLBaseStore(object):
                 0.0,
                 run_as_background_process,
                 "upsert_safety_check",
-                self._check_safe_to_upsert
+                self._check_safe_to_upsert,
             )
 
     @defer.inlineCallbacks
@@ -271,7 +272,7 @@ class SQLBaseStore(object):
                 15.0,
                 run_as_background_process,
                 "upsert_safety_check",
-                self._check_safe_to_upsert
+                self._check_safe_to_upsert,
             )
 
     def start_profiling(self):
@@ -298,13 +299,16 @@ class SQLBaseStore(object):
 
             perf_logger.info(
                 "Total database time: %.3f%% {%s} {%s}",
-                ratio * 100, top_three_counters, top_3_event_counters
+                ratio * 100,
+                top_three_counters,
+                top_3_event_counters,
             )
 
         self._clock.looping_call(loop, 10000)
 
-    def _new_transaction(self, conn, desc, after_callbacks, exception_callbacks,
-                         func, *args, **kwargs):
+    def _new_transaction(
+        self, conn, desc, after_callbacks, exception_callbacks, func, *args, **kwargs
+    ):
         start = time.time()
         txn_id = self._TXN_ID
 
@@ -312,7 +316,7 @@ class SQLBaseStore(object):
         # growing really large.
         self._TXN_ID = (self._TXN_ID + 1) % (MAX_TXN_ID)
 
-        name = "%s-%x" % (desc, txn_id, )
+        name = "%s-%x" % (desc, txn_id)
 
         transaction_logger.debug("[TXN START] {%s}", name)
 
@@ -323,7 +327,10 @@ class SQLBaseStore(object):
                 try:
                     txn = conn.cursor()
                     txn = LoggingTransaction(
-                        txn, name, self.database_engine, after_callbacks,
+                        txn,
+                        name,
+                        self.database_engine,
+                        after_callbacks,
                         exception_callbacks,
                     )
                     r = func(txn, *args, **kwargs)
@@ -334,7 +341,10 @@ class SQLBaseStore(object):
                     # transaction.
                     logger.warning(
                         "[TXN OPERROR] {%s} %s %d/%d",
-                        name, exception_to_unicode(e), i, N
+                        name,
+                        exception_to_unicode(e),
+                        i,
+                        N,
                     )
                     if i < N:
                         i += 1
@@ -342,8 +352,7 @@ class SQLBaseStore(object):
                             conn.rollback()
                         except self.database_engine.module.Error as e1:
                             logger.warning(
-                                "[TXN EROLL] {%s} %s",
-                                name, exception_to_unicode(e1),
+                                "[TXN EROLL] {%s} %s", name, exception_to_unicode(e1)
                             )
                         continue
                     raise
@@ -357,7 +366,8 @@ class SQLBaseStore(object):
                             except self.database_engine.module.Error as e1:
                                 logger.warning(
                                     "[TXN EROLL] {%s} %s",
-                                    name, exception_to_unicode(e1),
+                                    name,
+                                    exception_to_unicode(e1),
                                 )
                             continue
                     raise
@@ -396,16 +406,17 @@ class SQLBaseStore(object):
         exception_callbacks = []
 
         if LoggingContext.current_context() == LoggingContext.sentinel:
-            logger.warn(
-                "Starting db txn '%s' from sentinel context",
-                desc,
-            )
+            logger.warn("Starting db txn '%s' from sentinel context", desc)
 
         try:
             result = yield self.runWithConnection(
                 self._new_transaction,
-                desc, after_callbacks, exception_callbacks, func,
-                *args, **kwargs
+                desc,
+                after_callbacks,
+                exception_callbacks,
+                func,
+                *args,
+                **kwargs
             )
 
             for after_callback, after_args, after_kwargs in after_callbacks:
@@ -434,7 +445,7 @@ class SQLBaseStore(object):
         parent_context = LoggingContext.current_context()
         if parent_context == LoggingContext.sentinel:
             logger.warn(
-                "Starting db connection from sentinel context: metrics will be lost",
+                "Starting db connection from sentinel context: metrics will be lost"
             )
             parent_context = None
 
@@ -453,9 +464,7 @@ class SQLBaseStore(object):
                 return func(conn, *args, **kwargs)
 
         with PreserveLoggingContext():
-            result = yield self._db_pool.runWithConnection(
-                inner_func, *args, **kwargs
-            )
+            result = yield self._db_pool.runWithConnection(inner_func, *args, **kwargs)
 
         defer.returnValue(result)
 
@@ -469,9 +478,7 @@ class SQLBaseStore(object):
             A list of dicts where the key is the column header.
         """
         col_headers = list(intern(str(column[0])) for column in cursor.description)
-        results = list(
-            dict(zip(col_headers, row)) for row in cursor
-        )
+        results = list(dict(zip(col_headers, row)) for row in cursor)
         return results
 
     def _execute(self, desc, decoder, query, *args):
@@ -485,6 +492,7 @@ class SQLBaseStore(object):
         Returns:
             The result of decoder(results)
         """
+
         def interaction(txn):
             txn.execute(query, args)
             if decoder:
@@ -498,8 +506,7 @@ class SQLBaseStore(object):
     # no complex WHERE clauses, just a dict of values for columns.
 
     @defer.inlineCallbacks
-    def _simple_insert(self, table, values, or_ignore=False,
-                       desc="_simple_insert"):
+    def _simple_insert(self, table, values, or_ignore=False, desc="_simple_insert"):
         """Executes an INSERT query on the named table.
 
         Args:
@@ -511,10 +518,7 @@ class SQLBaseStore(object):
             `or_ignore` is True
         """
         try:
-            yield self.runInteraction(
-                desc,
-                self._simple_insert_txn, table, values,
-            )
+            yield self.runInteraction(desc, self._simple_insert_txn, table, values)
         except self.database_engine.module.IntegrityError:
             # We have to do or_ignore flag at this layer, since we can't reuse
             # a cursor after we receive an error from the db.
@@ -530,15 +534,13 @@ class SQLBaseStore(object):
         sql = "INSERT INTO %s (%s) VALUES(%s)" % (
             table,
             ", ".join(k for k in keys),
-            ", ".join("?" for _ in keys)
+            ", ".join("?" for _ in keys),
         )
 
         txn.execute(sql, vals)
 
     def _simple_insert_many(self, table, values, desc):
-        return self.runInteraction(
-            desc, self._simple_insert_many_txn, table, values
-        )
+        return self.runInteraction(desc, self._simple_insert_many_txn, table, values)
 
     @staticmethod
     def _simple_insert_many_txn(txn, table, values):
@@ -553,24 +555,18 @@ class SQLBaseStore(object):
         #
         # The sort is to ensure that we don't rely on dictionary iteration
         # order.
-        keys, vals = zip(*[
-            zip(
-                *(sorted(i.items(), key=lambda kv: kv[0]))
-            )
-            for i in values
-            if i
-        ])
+        keys, vals = zip(
+            *[zip(*(sorted(i.items(), key=lambda kv: kv[0]))) for i in values if i]
+        )
 
         for k in keys:
             if k != keys[0]:
-                raise RuntimeError(
-                    "All items must have the same keys"
-                )
+                raise RuntimeError("All items must have the same keys")
 
         sql = "INSERT INTO %s (%s) VALUES(%s)" % (
             table,
             ", ".join(k for k in keys[0]),
-            ", ".join("?" for _ in keys[0])
+            ", ".join("?" for _ in keys[0]),
         )
 
         txn.executemany(sql, vals)
@@ -583,7 +579,7 @@ class SQLBaseStore(object):
         values,
         insertion_values={},
         desc="_simple_upsert",
-        lock=True
+        lock=True,
     ):
         """
 
@@ -599,7 +595,7 @@ class SQLBaseStore(object):
 
         Args:
             table (str): The table to upsert into
-            keyvalues (dict): The unique key tables and their new values
+            keyvalues (dict): The unique key columns and their new values
             values (dict): The nonunique columns and their new values
             insertion_values (dict): additional key/values to use only when
                 inserting
@@ -631,17 +627,11 @@ class SQLBaseStore(object):
 
                 # presumably we raced with another transaction: let's retry.
                 logger.warn(
-                    "%s when upserting into %s; retrying: %s", e.__name__, table, e
+                    "IntegrityError when upserting into %s; retrying: %s", table, e
                 )
 
     def _simple_upsert_txn(
-        self,
-        txn,
-        table,
-        keyvalues,
-        values,
-        insertion_values={},
-        lock=True,
+        self, txn, table, keyvalues, values, insertion_values={}, lock=True
     ):
         """
         Pick the UPSERT method which works best on the platform. Either the
@@ -665,11 +655,7 @@ class SQLBaseStore(object):
             and table not in self._unsafe_to_upsert_tables
         ):
             return self._simple_upsert_txn_native_upsert(
-                txn,
-                table,
-                keyvalues,
-                values,
-                insertion_values=insertion_values,
+                txn, table, keyvalues, values, insertion_values=insertion_values
             )
         else:
             return self._simple_upsert_txn_emulated(
@@ -714,7 +700,7 @@ class SQLBaseStore(object):
             # SELECT instead to see if it exists.
             sql = "SELECT 1 FROM %s WHERE %s" % (
                 table,
-                " AND ".join(_getwhere(k) for k in keyvalues)
+                " AND ".join(_getwhere(k) for k in keyvalues),
             )
             sqlargs = list(keyvalues.values())
             txn.execute(sql, sqlargs)
@@ -726,7 +712,7 @@ class SQLBaseStore(object):
             sql = "UPDATE %s SET %s WHERE %s" % (
                 table,
                 ", ".join("%s = ?" % (k,) for k in values),
-                " AND ".join(_getwhere(k) for k in keyvalues)
+                " AND ".join(_getwhere(k) for k in keyvalues),
             )
             sqlargs = list(values.values()) + list(keyvalues.values())
 
@@ -773,19 +759,14 @@ class SQLBaseStore(object):
             latter = "NOTHING"
         else:
             allvalues.update(values)
-            latter = (
-                "UPDATE SET " + ", ".join(k + "=EXCLUDED." + k for k in values)
-            )
+            latter = "UPDATE SET " + ", ".join(k + "=EXCLUDED." + k for k in values)
 
-        sql = (
-            "INSERT INTO %s (%s) VALUES (%s) "
-            "ON CONFLICT (%s) DO %s"
-        ) % (
+        sql = ("INSERT INTO %s (%s) VALUES (%s) " "ON CONFLICT (%s) DO %s") % (
             table,
             ", ".join(k for k in allvalues),
             ", ".join("?" for _ in allvalues),
             ", ".join(k for k in keyvalues),
-            latter
+            latter,
         )
         txn.execute(sql, list(allvalues.values()))
 
@@ -870,8 +851,8 @@ class SQLBaseStore(object):
             latter = "NOTHING"
             value_values = [() for x in range(len(key_values))]
         else:
-            latter = (
-                "UPDATE SET " + ", ".join(k + "=EXCLUDED." + k for k in value_names)
+            latter = "UPDATE SET " + ", ".join(
+                k + "=EXCLUDED." + k for k in value_names
             )
 
         sql = "INSERT INTO %s (%s) VALUES (%s) ON CONFLICT (%s) DO %s" % (
@@ -889,8 +870,9 @@ class SQLBaseStore(object):
 
         return txn.execute_batch(sql, args)
 
-    def _simple_select_one(self, table, keyvalues, retcols,
-                           allow_none=False, desc="_simple_select_one"):
+    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
         return a single row, returning multiple columns from it.
 
@@ -903,14 +885,17 @@ class SQLBaseStore(object):
               statement returns no rows
         """
         return self.runInteraction(
-            desc,
-            self._simple_select_one_txn,
-            table, keyvalues, retcols, allow_none,
+            desc, self._simple_select_one_txn, table, keyvalues, retcols, allow_none
         )
 
-    def _simple_select_one_onecol(self, table, keyvalues, retcol,
-                                  allow_none=False,
-                                  desc="_simple_select_one_onecol"):
+    def _simple_select_one_onecol(
+        self,
+        table,
+        keyvalues,
+        retcol,
+        allow_none=False,
+        desc="_simple_select_one_onecol",
+    ):
         """Executes a SELECT query on the named table, which is expected to
         return a single row, returning a single column from it.
 
@@ -922,17 +907,18 @@ class SQLBaseStore(object):
         return self.runInteraction(
             desc,
             self._simple_select_one_onecol_txn,
-            table, keyvalues, retcol, allow_none=allow_none,
+            table,
+            keyvalues,
+            retcol,
+            allow_none=allow_none,
         )
 
     @classmethod
-    def _simple_select_one_onecol_txn(cls, txn, table, keyvalues, retcol,
-                                      allow_none=False):
+    def _simple_select_one_onecol_txn(
+        cls, txn, table, keyvalues, retcol, allow_none=False
+    ):
         ret = cls._simple_select_onecol_txn(
-            txn,
-            table=table,
-            keyvalues=keyvalues,
-            retcol=retcol,
+            txn, table=table, keyvalues=keyvalues, retcol=retcol
         )
 
         if ret:
@@ -945,12 +931,7 @@ class SQLBaseStore(object):
 
     @staticmethod
     def _simple_select_onecol_txn(txn, table, keyvalues, retcol):
-        sql = (
-            "SELECT %(retcol)s FROM %(table)s"
-        ) % {
-            "retcol": retcol,
-            "table": table,
-        }
+        sql = ("SELECT %(retcol)s FROM %(table)s") % {"retcol": retcol, "table": table}
 
         if keyvalues:
             sql += " WHERE %s" % " AND ".join("%s = ?" % k for k in iterkeys(keyvalues))
@@ -960,8 +941,9 @@ class SQLBaseStore(object):
 
         return [r[0] for r in txn]
 
-    def _simple_select_onecol(self, table, keyvalues, retcol,
-                              desc="_simple_select_onecol"):
+    def _simple_select_onecol(
+        self, table, keyvalues, retcol, desc="_simple_select_onecol"
+    ):
         """Executes a SELECT query on the named table, which returns a list
         comprising of the values of the named column from the selected rows.
 
@@ -974,13 +956,12 @@ class SQLBaseStore(object):
             Deferred: Results in a list
         """
         return self.runInteraction(
-            desc,
-            self._simple_select_onecol_txn,
-            table, keyvalues, retcol
+            desc, self._simple_select_onecol_txn, table, keyvalues, retcol
         )
 
-    def _simple_select_list(self, table, keyvalues, retcols,
-                            desc="_simple_select_list"):
+    def _simple_select_list(
+        self, table, keyvalues, retcols, desc="_simple_select_list"
+    ):
         """Executes a SELECT query on the named table, which may return zero or
         more rows, returning the result as a list of dicts.
 
@@ -994,9 +975,7 @@ class SQLBaseStore(object):
             defer.Deferred: resolves to list[dict[str, Any]]
         """
         return self.runInteraction(
-            desc,
-            self._simple_select_list_txn,
-            table, keyvalues, retcols
+            desc, self._simple_select_list_txn, table, keyvalues, retcols
         )
 
     @classmethod
@@ -1016,22 +995,26 @@ class SQLBaseStore(object):
             sql = "SELECT %s FROM %s WHERE %s" % (
                 ", ".join(retcols),
                 table,
-                " AND ".join("%s = ?" % (k, ) for k in keyvalues)
+                " AND ".join("%s = ?" % (k,) for k in keyvalues),
             )
             txn.execute(sql, list(keyvalues.values()))
         else:
-            sql = "SELECT %s FROM %s" % (
-                ", ".join(retcols),
-                table
-            )
+            sql = "SELECT %s FROM %s" % (", ".join(retcols), table)
             txn.execute(sql)
 
         return cls.cursor_to_dict(txn)
 
     @defer.inlineCallbacks
-    def _simple_select_many_batch(self, table, column, iterable, retcols,
-                                  keyvalues={}, desc="_simple_select_many_batch",
-                                  batch_size=100):
+    def _simple_select_many_batch(
+        self,
+        table,
+        column,
+        iterable,
+        retcols,
+        keyvalues={},
+        desc="_simple_select_many_batch",
+        batch_size=100,
+    ):
         """Executes a SELECT query on the named table, which may return zero or
         more rows, returning the result as a list of dicts.
 
@@ -1053,14 +1036,17 @@ class SQLBaseStore(object):
         it_list = list(iterable)
 
         chunks = [
-            it_list[i:i + batch_size]
-            for i in range(0, len(it_list), batch_size)
+            it_list[i : i + batch_size] for i in range(0, len(it_list), batch_size)
         ]
         for chunk in chunks:
             rows = yield self.runInteraction(
                 desc,
                 self._simple_select_many_txn,
-                table, column, chunk, keyvalues, retcols
+                table,
+                column,
+                chunk,
+                keyvalues,
+                retcols,
             )
 
             results.extend(rows)
@@ -1089,9 +1075,7 @@ class SQLBaseStore(object):
 
         clauses = []
         values = []
-        clauses.append(
-            "%s IN (%s)" % (column, ",".join("?" for _ in iterable))
-        )
+        clauses.append("%s IN (%s)" % (column, ",".join("?" for _ in iterable)))
         values.extend(iterable)
 
         for key, value in iteritems(keyvalues):
@@ -1099,19 +1083,14 @@ class SQLBaseStore(object):
             values.append(value)
 
         if clauses:
-            sql = "%s WHERE %s" % (
-                sql,
-                " AND ".join(clauses),
-            )
+            sql = "%s WHERE %s" % (sql, " AND ".join(clauses))
 
         txn.execute(sql, values)
         return cls.cursor_to_dict(txn)
 
     def _simple_update(self, table, keyvalues, updatevalues, desc):
         return self.runInteraction(
-            desc,
-            self._simple_update_txn,
-            table, keyvalues, updatevalues,
+            desc, self._simple_update_txn, table, keyvalues, updatevalues
         )
 
     @staticmethod
@@ -1127,15 +1106,13 @@ class SQLBaseStore(object):
             where,
         )
 
-        txn.execute(
-            update_sql,
-            list(updatevalues.values()) + list(keyvalues.values())
-        )
+        txn.execute(update_sql, list(updatevalues.values()) + list(keyvalues.values()))
 
         return txn.rowcount
 
-    def _simple_update_one(self, table, keyvalues, updatevalues,
-                           desc="_simple_update_one"):
+    def _simple_update_one(
+        self, table, keyvalues, updatevalues, desc="_simple_update_one"
+    ):
         """Executes an UPDATE query on the named table, setting new values for
         columns in a row matching the key values.
 
@@ -1154,9 +1131,7 @@ class SQLBaseStore(object):
         the update column in the 'keyvalues' dict as well.
         """
         return self.runInteraction(
-            desc,
-            self._simple_update_one_txn,
-            table, keyvalues, updatevalues,
+            desc, self._simple_update_one_txn, table, keyvalues, updatevalues
         )
 
     @classmethod
@@ -1169,12 +1144,11 @@ class SQLBaseStore(object):
             raise StoreError(500, "More than one row matched (%s)" % (table,))
 
     @staticmethod
-    def _simple_select_one_txn(txn, table, keyvalues, retcols,
-                               allow_none=False):
+    def _simple_select_one_txn(txn, table, keyvalues, retcols, allow_none=False):
         select_sql = "SELECT %s FROM %s WHERE %s" % (
             ", ".join(retcols),
             table,
-            " AND ".join("%s = ?" % (k,) for k in keyvalues)
+            " AND ".join("%s = ?" % (k,) for k in keyvalues),
         )
 
         txn.execute(select_sql, list(keyvalues.values()))
@@ -1197,9 +1171,7 @@ class SQLBaseStore(object):
             table : string giving the table name
             keyvalues : dict of column names and values to select the row with
         """
-        return self.runInteraction(
-            desc, self._simple_delete_one_txn, table, keyvalues
-        )
+        return self.runInteraction(desc, self._simple_delete_one_txn, table, keyvalues)
 
     @staticmethod
     def _simple_delete_one_txn(txn, table, keyvalues):
@@ -1212,7 +1184,7 @@ class SQLBaseStore(object):
         """
         sql = "DELETE FROM %s WHERE %s" % (
             table,
-            " AND ".join("%s = ?" % (k, ) for k in keyvalues)
+            " AND ".join("%s = ?" % (k,) for k in keyvalues),
         )
 
         txn.execute(sql, list(keyvalues.values()))
@@ -1222,15 +1194,13 @@ class SQLBaseStore(object):
             raise StoreError(500, "More than one row matched (%s)" % (table,))
 
     def _simple_delete(self, table, keyvalues, desc):
-        return self.runInteraction(
-            desc, self._simple_delete_txn, table, keyvalues
-        )
+        return self.runInteraction(desc, self._simple_delete_txn, table, keyvalues)
 
     @staticmethod
     def _simple_delete_txn(txn, table, keyvalues):
         sql = "DELETE FROM %s WHERE %s" % (
             table,
-            " AND ".join("%s = ?" % (k, ) for k in keyvalues)
+            " AND ".join("%s = ?" % (k,) for k in keyvalues),
         )
 
         return txn.execute(sql, list(keyvalues.values()))
@@ -1260,9 +1230,7 @@ class SQLBaseStore(object):
 
         clauses = []
         values = []
-        clauses.append(
-            "%s IN (%s)" % (column, ",".join("?" for _ in iterable))
-        )
+        clauses.append("%s IN (%s)" % (column, ",".join("?" for _ in iterable)))
         values.extend(iterable)
 
         for key, value in iteritems(keyvalues):
@@ -1270,14 +1238,12 @@ class SQLBaseStore(object):
             values.append(value)
 
         if clauses:
-            sql = "%s WHERE %s" % (
-                sql,
-                " AND ".join(clauses),
-            )
+            sql = "%s WHERE %s" % (sql, " AND ".join(clauses))
         return txn.execute(sql, values)
 
-    def _get_cache_dict(self, db_conn, table, entity_column, stream_column,
-                        max_value, limit=100000):
+    def _get_cache_dict(
+        self, db_conn, table, entity_column, stream_column, max_value, limit=100000
+    ):
         # Fetch a mapping of room_id -> max stream position for "recent" rooms.
         # It doesn't really matter how many we get, the StreamChangeCache will
         # do the right thing to ensure it respects the max size of cache.
@@ -1297,10 +1263,7 @@ class SQLBaseStore(object):
         txn = db_conn.cursor()
         txn.execute(sql, (int(max_value),))
 
-        cache = {
-            row[0]: int(row[1])
-            for row in txn
-        }
+        cache = {row[0]: int(row[1]) for row in txn}
 
         txn.close()
 
@@ -1342,9 +1305,7 @@ class SQLBaseStore(object):
         # be safe.
         for chunk in batch_iter(members_changed, 50):
             keys = itertools.chain([room_id], chunk)
-            self._send_invalidation_to_replication(
-                txn, _CURRENT_STATE_CACHE_NAME, keys,
-            )
+            self._send_invalidation_to_replication(txn, _CURRENT_STATE_CACHE_NAME, keys)
 
     def _invalidate_state_caches(self, room_id, members_changed):
         """Invalidates caches that are based on the current state, but does
@@ -1355,28 +1316,13 @@ class SQLBaseStore(object):
             members_changed (iterable[str]): The user_ids of members that have
                 changed
         """
-        for member in members_changed:
-            self._attempt_to_invalidate_cache(
-                "get_rooms_for_user_with_stream_ordering", (member,),
-            )
-
         for host in set(get_domain_from_id(u) for u in members_changed):
-            self._attempt_to_invalidate_cache(
-                "is_host_joined", (room_id, host,),
-            )
-            self._attempt_to_invalidate_cache(
-                "was_host_joined", (room_id, host,),
-            )
+            self._attempt_to_invalidate_cache("is_host_joined", (room_id, host))
+            self._attempt_to_invalidate_cache("was_host_joined", (room_id, host))
 
-        self._attempt_to_invalidate_cache(
-            "get_users_in_room", (room_id,),
-        )
-        self._attempt_to_invalidate_cache(
-            "get_room_summary", (room_id,),
-        )
-        self._attempt_to_invalidate_cache(
-            "get_current_state_ids", (room_id,),
-        )
+        self._attempt_to_invalidate_cache("get_users_in_room", (room_id,))
+        self._attempt_to_invalidate_cache("get_room_summary", (room_id,))
+        self._attempt_to_invalidate_cache("get_current_state_ids", (room_id,))
 
     def _attempt_to_invalidate_cache(self, cache_name, key):
         """Attempts to invalidate the cache of the given name, ignoring if the
@@ -1424,7 +1370,7 @@ class SQLBaseStore(object):
                     "cache_func": cache_name,
                     "keys": list(keys),
                     "invalidation_ts": self.clock.time_msec(),
-                }
+                },
             )
 
     def get_all_updated_caches(self, last_id, current_id, limit):
@@ -1440,11 +1386,10 @@ class SQLBaseStore(object):
                 " FROM cache_invalidation_stream"
                 " WHERE stream_id > ? ORDER BY stream_id ASC LIMIT ?"
             )
-            txn.execute(sql, (last_id, limit,))
+            txn.execute(sql, (last_id, limit))
             return txn.fetchall()
-        return self.runInteraction(
-            "get_all_updated_caches", get_all_updated_caches_txn
-        )
+
+        return self.runInteraction("get_all_updated_caches", get_all_updated_caches_txn)
 
     def get_cache_stream_token(self):
         if self._cache_id_gen:
@@ -1452,33 +1397,61 @@ class SQLBaseStore(object):
         else:
             return 0
 
-    def _simple_select_list_paginate(self, table, keyvalues, pagevalues, retcols,
-                                     desc="_simple_select_list_paginate"):
-        """Executes a SELECT query on the named table with start and limit,
+    def _simple_select_list_paginate(
+        self,
+        table,
+        keyvalues,
+        orderby,
+        start,
+        limit,
+        retcols,
+        order_direction="ASC",
+        desc="_simple_select_list_paginate",
+    ):
+        """
+        Executes a SELECT query on the named table with start and limit,
         of row numbers, which may return zero or number of rows from start to limit,
         returning the result as a list of dicts.
 
         Args:
             table (str): the table name
-            keyvalues (dict[str, Any] | None):
+            keyvalues (dict[str, T] | None):
                 column names and values to select the rows with, or None to not
                 apply a WHERE clause.
+            orderby (str): Column to order the results by.
+            start (int): Index to begin the query at.
+            limit (int): Number of results to return.
             retcols (iterable[str]): the names of the columns to return
-            order (str): order the select by this column
-            start (int): start number to begin the query from
-            limit (int): number of rows to reterive
+            order_direction (str): Whether the results should be ordered "ASC" or "DESC".
         Returns:
             defer.Deferred: resolves to list[dict[str, Any]]
         """
         return self.runInteraction(
             desc,
             self._simple_select_list_paginate_txn,
-            table, keyvalues, pagevalues, retcols
+            table,
+            keyvalues,
+            orderby,
+            start,
+            limit,
+            retcols,
+            order_direction=order_direction,
         )
 
     @classmethod
-    def _simple_select_list_paginate_txn(cls, txn, table, keyvalues, pagevalues, retcols):
-        """Executes a SELECT query on the named table with start and limit,
+    def _simple_select_list_paginate_txn(
+        cls,
+        txn,
+        table,
+        keyvalues,
+        orderby,
+        start,
+        limit,
+        retcols,
+        order_direction="ASC",
+    ):
+        """
+        Executes a SELECT query on the named table with start and limit,
         of row numbers, which may return zero or number of rows from start to limit,
         returning the result as a list of dicts.
 
@@ -1488,66 +1461,32 @@ class SQLBaseStore(object):
             keyvalues (dict[str, T] | None):
                 column names and values to select the rows with, or None to not
                 apply a WHERE clause.
-            pagevalues ([]):
-                order (str): order the select by this column
-                start (int): start number to begin the query from
-                limit (int): number of rows to reterive
+            orderby (str): Column to order the results by.
+            start (int): Index to begin the query at.
+            limit (int): Number of results to return.
             retcols (iterable[str]): the names of the columns to return
+            order_direction (str): Whether the results should be ordered "ASC" or "DESC".
         Returns:
             defer.Deferred: resolves to list[dict[str, Any]]
-
         """
+        if order_direction not in ["ASC", "DESC"]:
+            raise ValueError("order_direction must be one of 'ASC' or 'DESC'.")
+
         if keyvalues:
-            sql = "SELECT %s FROM %s WHERE %s ORDER BY %s" % (
-                ", ".join(retcols),
-                table,
-                " AND ".join("%s = ?" % (k,) for k in keyvalues),
-                " ? ASC LIMIT ? OFFSET ?"
-            )
-            txn.execute(sql, list(keyvalues.values()) + list(pagevalues))
+            where_clause = "WHERE " + " AND ".join("%s = ?" % (k,) for k in keyvalues)
         else:
-            sql = "SELECT %s FROM %s ORDER BY %s" % (
-                ", ".join(retcols),
-                table,
-                " ? ASC LIMIT ? OFFSET ?"
-            )
-            txn.execute(sql, pagevalues)
-
-        return cls.cursor_to_dict(txn)
+            where_clause = ""
 
-    @defer.inlineCallbacks
-    def get_user_list_paginate(self, table, keyvalues, pagevalues, retcols,
-                               desc="get_user_list_paginate"):
-        """Get a list of users from start row to a limit number of rows. This will
-        return a json object with users and total number of users in users list.
-
-        Args:
-            table (str): the table name
-            keyvalues (dict[str, Any] | None):
-                column names and values to select the rows with, or None to not
-                apply a WHERE clause.
-            pagevalues ([]):
-                order (str): order the select by this column
-                start (int): start number to begin the query from
-                limit (int): number of rows to reterive
-            retcols (iterable[str]): the names of the columns to return
-        Returns:
-            defer.Deferred: resolves to json object {list[dict[str, Any]], count}
-        """
-        users = yield self.runInteraction(
-            desc,
-            self._simple_select_list_paginate_txn,
-            table, keyvalues, pagevalues, retcols
-        )
-        count = yield self.runInteraction(
-            desc,
-            self.get_user_count_txn
+        sql = "SELECT %s FROM %s %s ORDER BY %s %s LIMIT ? OFFSET ?" % (
+            ", ".join(retcols),
+            table,
+            where_clause,
+            orderby,
+            order_direction,
         )
-        retval = {
-            "users": users,
-            "total": count
-        }
-        defer.returnValue(retval)
+        txn.execute(sql, list(keyvalues.values()) + [limit, start])
+
+        return cls.cursor_to_dict(txn)
 
     def get_user_count_txn(self, txn):
         """Get a total number of registered users in the users list.
@@ -1561,8 +1500,9 @@ class SQLBaseStore(object):
         txn.execute(sql_count)
         return txn.fetchone()[0]
 
-    def _simple_search_list(self, table, term, col, retcols,
-                            desc="_simple_search_list"):
+    def _simple_search_list(
+        self, table, term, col, retcols, desc="_simple_search_list"
+    ):
         """Executes a SELECT query on the named table, which may return zero or
         more rows, returning the result as a list of dicts.
 
@@ -1577,9 +1517,7 @@ class SQLBaseStore(object):
         """
 
         return self.runInteraction(
-            desc,
-            self._simple_search_list_txn,
-            table, term, col, retcols
+            desc, self._simple_search_list_txn, table, term, col, retcols
         )
 
     @classmethod
@@ -1598,11 +1536,7 @@ class SQLBaseStore(object):
             defer.Deferred: resolves to list[dict[str, Any]] or None
         """
         if term:
-            sql = "SELECT %s FROM %s WHERE %s LIKE ?" % (
-                ", ".join(retcols),
-                table,
-                col
-            )
+            sql = "SELECT %s FROM %s WHERE %s LIKE ?" % (", ".join(retcols), table, col)
             termvalues = ["%%" + term + "%%"]
             txn.execute(sql, termvalues)
         else:
@@ -1623,6 +1557,7 @@ class _RollbackButIsFineException(Exception):
     """ This exception is used to rollback a transaction without implying
     something went wrong.
     """
+
     pass