From cb8162d3d17a97574073d49bd6eef51c93b68157 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 20 Mar 2015 13:52:56 +0000
Subject: Rearrange storage modules

---
 synapse/storage/events.py | 387 ++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 387 insertions(+)
 create mode 100644 synapse/storage/events.py

(limited to 'synapse/storage/events.py')

diff --git a/synapse/storage/events.py b/synapse/storage/events.py
new file mode 100644
index 0000000000..b222dfb4aa
--- /dev/null
+++ b/synapse/storage/events.py
@@ -0,0 +1,387 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014, 2015 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.
+
+from _base import SQLBaseStore, _RollbackButIsFineException
+
+from twisted.internet import defer
+
+from synapse.util.logutils import log_function
+from synapse.api.constants import EventTypes
+from synapse.crypto.event_signing import compute_event_reference_hash
+
+from syutil.base64util import decode_base64
+from syutil.jsonutil import encode_canonical_json
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class EventsStore(SQLBaseStore):
+    @defer.inlineCallbacks
+    @log_function
+    def persist_event(self, event, context, backfilled=False,
+                      is_new_state=True, current_state=None):
+        stream_ordering = None
+        if backfilled:
+            if not self.min_token_deferred.called:
+                yield self.min_token_deferred
+            self.min_token -= 1
+            stream_ordering = self.min_token
+
+        try:
+            yield self.runInteraction(
+                "persist_event",
+                self._persist_event_txn,
+                event=event,
+                context=context,
+                backfilled=backfilled,
+                stream_ordering=stream_ordering,
+                is_new_state=is_new_state,
+                current_state=current_state,
+            )
+        except _RollbackButIsFineException:
+            pass
+
+    @defer.inlineCallbacks
+    def get_event(self, event_id, check_redacted=True,
+                  get_prev_content=False, allow_rejected=False,
+                  allow_none=False):
+        """Get an event from the database by event_id.
+
+        Args:
+            event_id (str): The event_id of the event to fetch
+            check_redacted (bool): If True, check if event has been redacted
+                and redact it.
+            get_prev_content (bool): If True and event is a state event,
+                include the previous states content in the unsigned field.
+            allow_rejected (bool): If True return rejected events.
+            allow_none (bool): If True, return None if no event found, if
+                False throw an exception.
+
+        Returns:
+            Deferred : A FrozenEvent.
+        """
+        event = yield self.runInteraction(
+            "get_event", self._get_event_txn,
+            event_id,
+            check_redacted=check_redacted,
+            get_prev_content=get_prev_content,
+            allow_rejected=allow_rejected,
+        )
+
+        if not event and not allow_none:
+            raise RuntimeError("Could not find event %s" % (event_id,))
+
+        defer.returnValue(event)
+
+    @log_function
+    def _persist_event_txn(self, txn, event, context, backfilled,
+                           stream_ordering=None, is_new_state=True,
+                           current_state=None):
+
+        # Remove the any existing cache entries for the event_id
+        self._get_event_cache.pop(event.event_id)
+
+        # We purposefully do this first since if we include a `current_state`
+        # key, we *want* to update the `current_state_events` table
+        if current_state:
+            txn.execute(
+                "DELETE FROM current_state_events WHERE room_id = ?",
+                (event.room_id,)
+            )
+
+            for s in current_state:
+                self._simple_insert_txn(
+                    txn,
+                    "current_state_events",
+                    {
+                        "event_id": s.event_id,
+                        "room_id": s.room_id,
+                        "type": s.type,
+                        "state_key": s.state_key,
+                    },
+                )
+
+        if event.is_state() and is_new_state:
+            if not backfilled and not context.rejected:
+                self._simple_insert_txn(
+                    txn,
+                    table="state_forward_extremities",
+                    values={
+                        "event_id": event.event_id,
+                        "room_id": event.room_id,
+                        "type": event.type,
+                        "state_key": event.state_key,
+                    },
+                )
+
+                for prev_state_id, _ in event.prev_state:
+                    self._simple_delete_txn(
+                        txn,
+                        table="state_forward_extremities",
+                        keyvalues={
+                            "event_id": prev_state_id,
+                        }
+                    )
+
+        outlier = event.internal_metadata.is_outlier()
+
+        if not outlier:
+            self._store_state_groups_txn(txn, event, context)
+
+            self._update_min_depth_for_room_txn(
+                txn,
+                event.room_id,
+                event.depth
+            )
+
+        self._handle_prev_events(
+            txn,
+            outlier=outlier,
+            event_id=event.event_id,
+            prev_events=event.prev_events,
+            room_id=event.room_id,
+        )
+
+        have_persisted = self._simple_select_one_onecol_txn(
+            txn,
+            table="event_json",
+            keyvalues={"event_id": event.event_id},
+            retcol="event_id",
+            allow_none=True,
+        )
+
+        metadata_json = encode_canonical_json(
+            event.internal_metadata.get_dict()
+        )
+
+        # If we have already persisted this event, we don't need to do any
+        # more processing.
+        # The processing above must be done on every call to persist event,
+        # since they might not have happened on previous calls. For example,
+        # if we are persisting an event that we had persisted as an outlier,
+        # but is no longer one.
+        if have_persisted:
+            if not outlier:
+                sql = (
+                    "UPDATE event_json SET internal_metadata = ?"
+                    " WHERE event_id = ?"
+                )
+                txn.execute(
+                    sql,
+                    (metadata_json.decode("UTF-8"), event.event_id,)
+                )
+
+                sql = (
+                    "UPDATE events SET outlier = 0"
+                    " WHERE event_id = ?"
+                )
+                txn.execute(
+                    sql,
+                    (event.event_id,)
+                )
+            return
+
+        if event.type == EventTypes.Member:
+            self._store_room_member_txn(txn, event)
+        elif event.type == EventTypes.Name:
+            self._store_room_name_txn(txn, event)
+        elif event.type == EventTypes.Topic:
+            self._store_room_topic_txn(txn, event)
+        elif event.type == EventTypes.Redaction:
+            self._store_redaction(txn, event)
+
+        event_dict = {
+            k: v
+            for k, v in event.get_dict().items()
+            if k not in [
+                "redacted",
+                "redacted_because",
+            ]
+        }
+
+        self._simple_insert_txn(
+            txn,
+            table="event_json",
+            values={
+                "event_id": event.event_id,
+                "room_id": event.room_id,
+                "internal_metadata": metadata_json.decode("UTF-8"),
+                "json": encode_canonical_json(event_dict).decode("UTF-8"),
+            },
+        )
+
+        content = encode_canonical_json(
+            event.content
+        ).decode("UTF-8")
+
+        vals = {
+            "topological_ordering": event.depth,
+            "event_id": event.event_id,
+            "type": event.type,
+            "room_id": event.room_id,
+            "content": content,
+            "processed": True,
+            "outlier": outlier,
+            "depth": event.depth,
+        }
+
+        if stream_ordering is None:
+            stream_ordering = self.get_next_stream_id()
+
+
+        unrec = {
+            k: v
+            for k, v in event.get_dict().items()
+            if k not in vals.keys() and k not in [
+                "redacted",
+                "redacted_because",
+                "signatures",
+                "hashes",
+                "prev_events",
+            ]
+        }
+
+        vals["unrecognized_keys"] = encode_canonical_json(
+            unrec
+        ).decode("UTF-8")
+
+        sql = (
+            "INSERT INTO events"
+            " (stream_ordering, topological_ordering, event_id, type,"
+            " room_id, content, processed, outlier, depth)"
+            " VALUES (%s,?,?,?,?,?,?,?,?)"
+        ) % (stream_ordering,)
+
+        txn.execute(
+            sql,
+            (event.depth, event.event_id, event.type, event.room_id,
+             content, True, outlier, event.depth)
+        )
+
+        if context.rejected:
+            self._store_rejections_txn(txn, event.event_id, context.rejected)
+
+        if event.is_state():
+            vals = {
+                "event_id": event.event_id,
+                "room_id": event.room_id,
+                "type": event.type,
+                "state_key": event.state_key,
+            }
+
+            # TODO: How does this work with backfilling?
+            if hasattr(event, "replaces_state"):
+                vals["prev_state"] = event.replaces_state
+
+            self._simple_insert_txn(
+                txn,
+                "state_events",
+                vals,
+            )
+
+            if is_new_state and not context.rejected:
+                self._simple_insert_txn(
+                    txn,
+                    "current_state_events",
+                    {
+                        "event_id": event.event_id,
+                        "room_id": event.room_id,
+                        "type": event.type,
+                        "state_key": event.state_key,
+                    },
+                )
+
+            for e_id, h in event.prev_state:
+                self._simple_insert_txn(
+                    txn,
+                    table="event_edges",
+                    values={
+                        "event_id": event.event_id,
+                        "prev_event_id": e_id,
+                        "room_id": event.room_id,
+                        "is_state": 1,
+                    },
+                )
+
+        for hash_alg, hash_base64 in event.hashes.items():
+            hash_bytes = decode_base64(hash_base64)
+            self._store_event_content_hash_txn(
+                txn, event.event_id, hash_alg, hash_bytes,
+            )
+
+        for prev_event_id, prev_hashes in event.prev_events:
+            for alg, hash_base64 in prev_hashes.items():
+                hash_bytes = decode_base64(hash_base64)
+                self._store_prev_event_hash_txn(
+                    txn, event.event_id, prev_event_id, alg, hash_bytes
+                )
+
+        for auth_id, _ in event.auth_events:
+            self._simple_insert_txn(
+                txn,
+                table="event_auth",
+                values={
+                    "event_id": event.event_id,
+                    "room_id": event.room_id,
+                    "auth_id": auth_id,
+                },
+            )
+
+        (ref_alg, ref_hash_bytes) = compute_event_reference_hash(event)
+        self._store_event_reference_hash_txn(
+            txn, event.event_id, ref_alg, ref_hash_bytes
+        )
+
+    def _store_redaction(self, txn, event):
+        # invalidate the cache for the redacted event
+        self._get_event_cache.pop(event.redacts)
+        txn.execute(
+            "INSERT INTO redactions (event_id, redacts) VALUES (?,?)",
+            (event.event_id, event.redacts)
+        )
+
+    def have_events(self, event_ids):
+        """Given a list of event ids, check if we have already processed them.
+
+        Returns:
+            dict: Has an entry for each event id we already have seen. Maps to
+            the rejected reason string if we rejected the event, else maps to
+            None.
+        """
+        if not event_ids:
+            return defer.succeed({})
+
+        def f(txn):
+            sql = (
+                "SELECT e.event_id, reason FROM events as e "
+                "LEFT JOIN rejections as r ON e.event_id = r.event_id "
+                "WHERE e.event_id = ?"
+            )
+
+            res = {}
+            for event_id in event_ids:
+                txn.execute(sql, (event_id,))
+                row = txn.fetchone()
+                if row:
+                    _, rejected = row
+                    res[event_id] = rejected
+
+            return res
+
+        return self.runInteraction(
+            "have_events", f,
+        )
-- 
cgit 1.4.1


From f488293d96dc690ce7a4b3163fe8c4e0d3e31899 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Tue, 24 Mar 2015 16:20:26 +0000
Subject: Don't reinsert into event_edges

---
 synapse/storage/events.py | 16 ++++++++--------
 1 file changed, 8 insertions(+), 8 deletions(-)

(limited to 'synapse/storage/events.py')

diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 542559c59f..4d636d3f46 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -149,14 +149,6 @@ class EventsStore(SQLBaseStore):
                 event.depth
             )
 
-        self._handle_prev_events(
-            txn,
-            outlier=outlier,
-            event_id=event.event_id,
-            prev_events=event.prev_events,
-            room_id=event.room_id,
-        )
-
         have_persisted = self._simple_select_one_onecol_txn(
             txn,
             table="event_json",
@@ -196,6 +188,14 @@ class EventsStore(SQLBaseStore):
                 )
             return
 
+        self._handle_prev_events(
+            txn,
+            outlier=outlier,
+            event_id=event.event_id,
+            prev_events=event.prev_events,
+            room_id=event.room_id,
+        )
+
         if event.type == EventTypes.Member:
             self._store_room_member_txn(txn, event)
         elif event.type == EventTypes.Name:
-- 
cgit 1.4.1


From 0e8f5095c7e7075b249ad53a9f60a4d2fdeeaaed Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Wed, 25 Mar 2015 17:15:20 +0000
Subject: Fix unicode database support

---
 synapse/app/homeserver.py                          | 47 +++++++++++++---------
 synapse/handlers/login.py                          |  2 +-
 synapse/rest/client/v1/profile.py                  |  7 +++-
 synapse/storage/__init__.py                        | 12 ++++--
 synapse/storage/_base.py                           |  4 ++
 synapse/storage/events.py                          |  8 ++--
 synapse/storage/keys.py                            |  4 +-
 synapse/storage/profile.py                         | 12 +++++-
 synapse/storage/registration.py                    | 18 +++++++--
 synapse/storage/room.py                            |  1 +
 .../schema/full_schemas/11/media_repository.sql    |  2 +-
 .../storage/schema/full_schemas/11/profiles.sql    |  2 +-
 .../schema/full_schemas/11/transactions.sql        |  1 -
 synapse/storage/signatures.py                      | 10 ++---
 synapse/storage/transactions.py                    |  2 +-
 15 files changed, 88 insertions(+), 44 deletions(-)

(limited to 'synapse/storage/events.py')

diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py
index 394e93e6c2..beab6ffc7a 100755
--- a/synapse/app/homeserver.py
+++ b/synapse/app/homeserver.py
@@ -110,14 +110,12 @@ class SynapseHomeServer(HomeServer):
             return None
 
     def build_db_pool(self):
-        name = self.db_config.pop("name", None)
-        if name == "MySQLdb":
-            return adbapi.ConnectionPool(
-                name,
-                **self.db_config
-            )
+        name = self.db_config["name"]
 
-        raise RuntimeError("Unsupported database type")
+        return adbapi.ConnectionPool(
+            name,
+            **self.db_config.get("args", {})
+        )
 
     def create_resource_tree(self, redirect_root_to_web_client):
         """Create the resource tree for this Home Server.
@@ -323,7 +321,7 @@ def change_resource_limit(soft_file_no):
         resource.setrlimit(resource.RLIMIT_NOFILE, (soft_file_no, hard))
 
         logger.info("Set file limit to: %d", soft_file_no)
-    except (ValueError, resource.error) as e:
+    except (   ValueError, resource.error) as e:
         logger.warn("Failed to set file limit: %s", e)
 
 
@@ -363,20 +361,33 @@ def setup(config_options):
     if config.database_config:
         with open(config.database_config, 'r') as f:
             db_config = yaml.safe_load(f)
-
-        name = db_config.get("name", None)
-        if name == "MySQLdb":
-            db_config.update({
-                "sql_mode": "TRADITIONAL",
-                "charset": "utf8",
-                "use_unicode": True,
-            })
     else:
         db_config = {
             "name": "sqlite3",
             "database": config.database_path,
         }
 
+    db_config = {
+        k: v for k, v in db_config.items()
+        if not k.startswith("cp_")
+    }
+
+    name = db_config.get("name", None)
+    if name in ["MySQLdb", "mysql.connector"]:
+        db_config.setdefault("args", {}).update({
+            "sql_mode": "TRADITIONAL",
+            "charset": "utf8",
+            "use_unicode": True,
+        })
+    elif name == "sqlite3":
+        db_config.setdefault("args", {}).update({
+            "cp_min": 1,
+            "cp_max": 1,
+            "cp_openfun": prepare_database,
+        })
+    else:
+        raise RuntimeError("Unsupported database type '%s'" % (name,))
+
     hs = SynapseHomeServer(
         config.server_name,
         domain_with_port=domain_with_port,
@@ -401,8 +412,8 @@ def setup(config_options):
         # with sqlite3.connect(db_name) as db_conn:
         #     prepare_sqlite3_database(db_conn)
         #     prepare_database(db_conn)
-        import MySQLdb
-        db_conn = MySQLdb.connect(**db_config)
+        import mysql.connector
+        db_conn = mysql.connector.connect(**db_config.get("args", {}))
         prepare_database(db_conn)
     except UpgradeDatabaseException:
         sys.stderr.write(
diff --git a/synapse/handlers/login.py b/synapse/handlers/login.py
index 7447800460..76647c7941 100644
--- a/synapse/handlers/login.py
+++ b/synapse/handlers/login.py
@@ -57,7 +57,7 @@ class LoginHandler(BaseHandler):
             logger.warn("Attempted to login as %s but they do not exist", user)
             raise LoginError(403, "", errcode=Codes.FORBIDDEN)
 
-        stored_hash = user_info[0]["password_hash"]
+        stored_hash = user_info["password_hash"]
         if bcrypt.checkpw(password, stored_hash):
             # generate an access token and store it.
             token = self.reg_handler._generate_token(user)
diff --git a/synapse/rest/client/v1/profile.py b/synapse/rest/client/v1/profile.py
index 1e77eb49cf..7387b4adb9 100644
--- a/synapse/rest/client/v1/profile.py
+++ b/synapse/rest/client/v1/profile.py
@@ -19,9 +19,13 @@ from twisted.internet import defer
 from .base import ClientV1RestServlet, client_path_pattern
 from synapse.types import UserID
 
+import logging
 import simplejson as json
 
 
+logger = logging.getLogger(__name__)
+
+
 class ProfileDisplaynameRestServlet(ClientV1RestServlet):
     PATTERN = client_path_pattern("/profile/(?P<user_id>[^/]*)/displayname")
 
@@ -47,7 +51,8 @@ class ProfileDisplaynameRestServlet(ClientV1RestServlet):
             defer.returnValue((400, "Unable to parse name"))
 
         yield self.handlers.profile_handler.set_displayname(
-            user, auth_user, new_name)
+            user, auth_user, new_name
+        )
 
         defer.returnValue((200, {}))
 
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index e03d55b00d..abde7d0df5 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -410,10 +410,14 @@ def executescript(txn, schema_path):
 
 
 def _get_or_create_schema_state(txn):
-    schema_path = os.path.join(
-        dir_path, "schema", "schema_version.sql",
-    )
-    executescript(txn, schema_path)
+    try:
+        # Bluntly try creating the schema_version tables.
+        schema_path = os.path.join(
+            dir_path, "schema", "schema_version.sql",
+        )
+        executescript(txn, schema_path)
+    except:
+        pass
 
     txn.execute("SELECT version, upgraded FROM schema_version")
     row = txn.fetchone()
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 1ea39bc0ad..76ec3ee93f 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -755,6 +755,8 @@ class SQLBaseStore(object):
             return None
 
         internal_metadata, js, redacted, rejected_reason = res
+        js = js.decode("utf8")
+        internal_metadata = internal_metadata.decode("utf8")
 
         start_time = update_counter("select_event", start_time)
 
@@ -779,9 +781,11 @@ class SQLBaseStore(object):
             sql_getevents_timer.inc_by(curr_time - last_time, desc)
             return curr_time
 
+        logger.debug("Got js: %r", js)
         d = json.loads(js)
         start_time = update_counter("decode_json", start_time)
 
+        logger.debug("Got internal_metadata: %r", internal_metadata)
         internal_metadata = json.loads(internal_metadata)
         start_time = update_counter("decode_internal", start_time)
 
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 4d636d3f46..69f598967e 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -294,15 +294,17 @@ class EventsStore(SQLBaseStore):
             )
 
             if is_new_state and not context.rejected:
-                self._simple_insert_txn(
+                self._simple_upsert_txn(
                     txn,
                     "current_state_events",
-                    {
-                        "event_id": event.event_id,
+                    keyvalues={
                         "room_id": event.room_id,
                         "type": event.type,
                         "state_key": event.state_key,
                     },
+                    values={
+                        "event_id": event.event_id,
+                    }
                 )
 
             for e_id, h in event.prev_state:
diff --git a/synapse/storage/keys.py b/synapse/storage/keys.py
index 25fef79434..e6975a945b 100644
--- a/synapse/storage/keys.py
+++ b/synapse/storage/keys.py
@@ -64,7 +64,7 @@ class KeyStore(SQLBaseStore):
                 "fingerprint": fingerprint,
                 "from_server": from_server,
                 "ts_added_ms": time_now_ms,
-                "tls_certificate": buffer(tls_certificate_bytes),
+                "tls_certificate": tls_certificate_bytes,
             },
         )
 
@@ -113,6 +113,6 @@ class KeyStore(SQLBaseStore):
                 "key_id": "%s:%s" % (verify_key.alg, verify_key.version),
                 "from_server": from_server,
                 "ts_added_ms": time_now_ms,
-                "verify_key": buffer(verify_key.encode()),
+                "verify_key": verify_key.encode(),
             },
         )
diff --git a/synapse/storage/profile.py b/synapse/storage/profile.py
index a6e52cb248..09778045bf 100644
--- a/synapse/storage/profile.py
+++ b/synapse/storage/profile.py
@@ -13,6 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from twisted.internet import defer
+
 from ._base import SQLBaseStore
 
 
@@ -24,19 +26,25 @@ class ProfileStore(SQLBaseStore):
             desc="create_profile",
         )
 
+    @defer.inlineCallbacks
     def get_profile_displayname(self, user_localpart):
-        return self._simple_select_one_onecol(
+        name = yield self._simple_select_one_onecol(
             table="profiles",
             keyvalues={"user_id": user_localpart},
             retcol="displayname",
             desc="get_profile_displayname",
         )
 
+        if name:
+            name = name.decode("utf8")
+
+        defer.returnValue(name)
+
     def set_profile_displayname(self, user_localpart, new_displayname):
         return self._simple_update_one(
             table="profiles",
             keyvalues={"user_id": user_localpart},
-            updatevalues={"displayname": new_displayname},
+            updatevalues={"displayname": new_displayname.encode("utf8")},
             desc="set_profile_displayname",
         )
 
diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py
index fe26d6d62f..7258f7b2a5 100644
--- a/synapse/storage/registration.py
+++ b/synapse/storage/registration.py
@@ -81,13 +81,23 @@ class RegistrationStore(SQLBaseStore):
         txn.execute("INSERT INTO access_tokens(user_id, token) " +
                     "VALUES (?,?)", [user_id, token])
 
+    @defer.inlineCallbacks
     def get_user_by_id(self, user_id):
-        query = ("SELECT users.name, users.password_hash FROM users"
-                 " WHERE users.name = ?")
-        return self._execute(
-            "get_user_by_id", self.cursor_to_dict, query, user_id
+        user_info = yield self._simple_select_one(
+            table="users",
+            keyvalues={
+                "name": user_id,
+            },
+            retcols=["name", "password_hash"],
+            allow_none=True,
         )
 
+        if user_info:
+            user_info["password_hash"] = user_info["password_hash"].decode("utf8")
+
+        defer.returnValue(user_info)
+
+
     @cached()
     # TODO(paul): Currently there's no code to invalidate this cache. That
     #   means if/when we ever add internal ways to invalidate access tokens or
diff --git a/synapse/storage/room.py b/synapse/storage/room.py
index 501e947ad7..a1a76280fe 100644
--- a/synapse/storage/room.py
+++ b/synapse/storage/room.py
@@ -72,6 +72,7 @@ class RoomStore(SQLBaseStore):
             keyvalues={"room_id": room_id},
             retcols=RoomsTable.fields,
             desc="get_room",
+            allow_none=True,
         )
 
     @defer.inlineCallbacks
diff --git a/synapse/storage/schema/full_schemas/11/media_repository.sql b/synapse/storage/schema/full_schemas/11/media_repository.sql
index 8bc84dc24d..d9559f5902 100644
--- a/synapse/storage/schema/full_schemas/11/media_repository.sql
+++ b/synapse/storage/schema/full_schemas/11/media_repository.sql
@@ -65,4 +65,4 @@ CREATE TABLE IF NOT EXISTS remote_media_cache_thumbnails (
 ) ENGINE = INNODB;
 
 CREATE INDEX IF NOT EXISTS remote_media_cache_thumbnails_media_id
-    ON local_media_repository_thumbnails (media_id);
+    ON remote_media_cache_thumbnails (media_id);
diff --git a/synapse/storage/schema/full_schemas/11/profiles.sql b/synapse/storage/schema/full_schemas/11/profiles.sql
index 32defe2f79..552645c56f 100644
--- a/synapse/storage/schema/full_schemas/11/profiles.sql
+++ b/synapse/storage/schema/full_schemas/11/profiles.sql
@@ -14,7 +14,7 @@
  */
 CREATE TABLE IF NOT EXISTS profiles(
     user_id VARCHAR(255) NOT NULL,
-    displayname VARCHAR(255),
+    displayname VARBINARY(255),
     avatar_url VARCHAR(255),
     UNIQUE(user_id)
 ) ENGINE = INNODB;
diff --git a/synapse/storage/schema/full_schemas/11/transactions.sql b/synapse/storage/schema/full_schemas/11/transactions.sql
index 0570bf95d9..bd13bba8c2 100644
--- a/synapse/storage/schema/full_schemas/11/transactions.sql
+++ b/synapse/storage/schema/full_schemas/11/transactions.sql
@@ -38,7 +38,6 @@ CREATE TABLE IF NOT EXISTS sent_transactions(
 ) ENGINE = INNODB;
 
 CREATE INDEX IF NOT EXISTS sent_transaction_dest ON sent_transactions(destination);
-CREATE INDEX IF NOT EXISTS sent_transaction_dest_referenced ON sent_transactions(destination);
 CREATE INDEX IF NOT EXISTS sent_transaction_txn_id ON sent_transactions(transaction_id);
 -- So that we can do an efficient look up of all transactions that have yet to be successfully
 -- sent.
diff --git a/synapse/storage/signatures.py b/synapse/storage/signatures.py
index 13ce335101..35bba854f9 100644
--- a/synapse/storage/signatures.py
+++ b/synapse/storage/signatures.py
@@ -54,7 +54,7 @@ class SignatureStore(SQLBaseStore):
             {
                 "event_id": event_id,
                 "algorithm": algorithm,
-                "hash": buffer(hash_bytes),
+                "hash": hash_bytes,
             },
         )
 
@@ -99,7 +99,7 @@ class SignatureStore(SQLBaseStore):
             " WHERE event_id = ?"
         )
         txn.execute(query, (event_id, ))
-        return dict(txn.fetchall())
+        return {k: v for k, v in txn.fetchall()}
 
     def _store_event_reference_hash_txn(self, txn, event_id, algorithm,
                                         hash_bytes):
@@ -116,7 +116,7 @@ class SignatureStore(SQLBaseStore):
             {
                 "event_id": event_id,
                 "algorithm": algorithm,
-                "hash": buffer(hash_bytes),
+                "hash": hash_bytes,
             },
         )
 
@@ -160,7 +160,7 @@ class SignatureStore(SQLBaseStore):
                 "event_id": event_id,
                 "signature_name": signature_name,
                 "key_id": key_id,
-                "signature": buffer(signature_bytes),
+                "signature": signature_bytes,
             },
         )
 
@@ -193,6 +193,6 @@ class SignatureStore(SQLBaseStore):
                 "event_id": event_id,
                 "prev_event_id": prev_event_id,
                 "algorithm": algorithm,
-                "hash": buffer(hash_bytes),
+                "hash": hash_bytes,
             },
         )
diff --git a/synapse/storage/transactions.py b/synapse/storage/transactions.py
index 03e1e3b808..e3e484fb2d 100644
--- a/synapse/storage/transactions.py
+++ b/synapse/storage/transactions.py
@@ -282,7 +282,7 @@ class TransactionStore(SQLBaseStore):
         query = (
             "UPDATE destinations"
             " SET retry_last_ts = ?, retry_interval = ?"
-            " WHERE destinations = ?"
+            " WHERE destination = ?"
         )
 
         txn.execute(
-- 
cgit 1.4.1


From 304111afd0fdda6a0c58d81238a1bbfa7f318208 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Tue, 7 Apr 2015 12:05:36 +0100
Subject: Don't use AUTOINCREMENT, use an in memory version

---
 synapse/storage/_base.py        | 54 ++++++++++++++++++++++++++++++++++++-----
 synapse/storage/events.py       | 13 +++++-----
 synapse/storage/registration.py | 10 +++++---
 synapse/storage/state.py        |  6 ++---
 synapse/storage/transactions.py |  4 ++-
 5 files changed, 68 insertions(+), 19 deletions(-)

(limited to 'synapse/storage/events.py')

diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index de4f661973..9f63f07080 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -163,8 +163,8 @@ class LoggingTransaction(object):
             return self.txn.execute(
                 sql, *args, **kwargs
             )
-        except:
-                logger.exception("[SQL FAIL] {%s}", self.name)
+        except Exception as e:
+                logger.debug("[SQL FAIL] {%s} %s", self.name, e)
                 raise
         finally:
             msecs = (time.time() * 1000) - start
@@ -209,6 +209,46 @@ class PerformanceCounters(object):
         return top_n_counters
 
 
+class IdGenerator(object):
+    def __init__(self, table, column, store):
+        self.table = table
+        self.column = column
+        self.store = store
+        self._lock = threading.Lock()
+        self._next_id = None
+
+    @defer.inlineCallbacks
+    def get_next(self):
+        with self._lock:
+            if not self._next_id:
+                res = yield self.store._execute_and_decode(
+                    "IdGenerator_%s" % (self.table,),
+                    "SELECT MAX(%s) as mx FROM %s" % (self.column, self.table,)
+                )
+
+                self._next_id = (res and res[0] and res[0]["mx"]) or 1
+
+            i = self._next_id
+            self._next_id += 1
+            defer.returnValue(i)
+
+    def get_next_txn(self, txn):
+        with self._lock:
+            if self._next_id:
+                i = self._next_id
+                self._next_id += 1
+                return i
+            else:
+                txn.execute(
+                    "SELECT MAX(%s) FROM %s" % (self.column, self.table,)
+                )
+
+                val, = txn.fetchone()
+                self._next_id = val or 2
+
+                return 1
+
+
 class SQLBaseStore(object):
     _TXN_ID = 0
 
@@ -234,8 +274,10 @@ class SQLBaseStore(object):
         # Pretend the getEventCache is just another named cache
         caches_by_name["*getEvent*"] = self._get_event_cache
 
-        self._next_stream_id_lock = threading.Lock()
-        self._next_stream_id = int(hs.get_clock().time_msec()) * 1000
+        self._stream_id_gen = IdGenerator("events", "stream_ordering", self)
+        self._transaction_id_gen = IdGenerator("sent_transactions", "id", self)
+        self._state_groups_id_gen = IdGenerator("state_groups", "id", self)
+        self._access_tokens_id_gen = IdGenerator("access_tokens", "id", self)
 
     def start_profiling(self):
         self._previous_loop_ts = self._clock.time_msec()
@@ -292,8 +334,8 @@ class SQLBaseStore(object):
                         LoggingTransaction(txn, name, self.database_engine),
                         *args, **kwargs
                     )
-                except:
-                    logger.exception("[TXN FAIL] {%s}", name)
+                except Exception as e:
+                    logger.debug("[TXN FAIL] {%s}", name, e)
                     raise
                 finally:
                     end = time.time() * 1000
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 69f598967e..514feebcbf 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -96,12 +96,16 @@ class EventsStore(SQLBaseStore):
         # Remove the any existing cache entries for the event_id
         self._get_event_cache.pop(event.event_id)
 
+        if stream_ordering is None:
+            stream_ordering = self._stream_id_gen.get_next_txn(txn)
+
         # We purposefully do this first since if we include a `current_state`
         # key, we *want* to update the `current_state_events` table
         if current_state:
-            txn.execute(
-                "DELETE FROM current_state_events WHERE room_id = ?",
-                (event.room_id,)
+            self._simple_delete_txn(
+                txn,
+                table="current_state_events",
+                keyvalues={"room_id": event.room_id},
             )
 
             for s in current_state:
@@ -240,9 +244,6 @@ class EventsStore(SQLBaseStore):
             "depth": event.depth,
         }
 
-        if stream_ordering is None:
-            stream_ordering = self.get_next_stream_id()
-
         unrec = {
             k: v
             for k, v in event.get_dict().items()
diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py
index 0c785ec989..b62b4a3414 100644
--- a/synapse/storage/registration.py
+++ b/synapse/storage/registration.py
@@ -39,10 +39,12 @@ class RegistrationStore(SQLBaseStore):
         Raises:
             StoreError if there was a problem adding this.
         """
-        yield self._simple_insert(
+        next_id = yield self._access_tokens_id_gen.get_next()
+
+        self._simple_insert(
             "access_tokens",
             {
-                "id": self.get_next_stream_id(),
+                "id": next_id,
                 "user_id": user_id,
                 "token": token
             },
@@ -68,6 +70,8 @@ class RegistrationStore(SQLBaseStore):
     def _register(self, txn, user_id, token, password_hash):
         now = int(self.clock.time())
 
+        next_id = self._access_tokens_id_gen.get_next_txn(txn)
+
         try:
             txn.execute("INSERT INTO users(name, password_hash, creation_ts) "
                         "VALUES (?,?,?)",
@@ -82,7 +86,7 @@ class RegistrationStore(SQLBaseStore):
         txn.execute(
             "INSERT INTO access_tokens(id, user_id, token)"
             " VALUES (?,?,?)",
-            (self.get_next_stream_id(), user_id, token,)
+            (next_id, user_id, token,)
         )
 
     @defer.inlineCallbacks
diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index 65ea9c4d83..3e55cb81bf 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -93,12 +93,12 @@ class StateStore(SQLBaseStore):
 
         state_group = context.state_group
         if not state_group:
-            group = _make_group_id(self._clock)
-            state_group = self._simple_insert_txn(
+            state_group = _make_group_id(self._clock)
+            self._simple_insert_txn(
                 txn,
                 table="state_groups",
                 values={
-                    "id": group,
+                    "id": state_group,
                     "room_id": event.room_id,
                     "event_id": event.event_id,
                 },
diff --git a/synapse/storage/transactions.py b/synapse/storage/transactions.py
index e3e484fb2d..9594fe1f2b 100644
--- a/synapse/storage/transactions.py
+++ b/synapse/storage/transactions.py
@@ -123,6 +123,8 @@ class TransactionStore(SQLBaseStore):
     def _prep_send_transaction(self, txn, transaction_id, destination,
                                origin_server_ts):
 
+        next_id = self._transaction_id_gen.get_next_txn(txn)
+
         # First we find out what the prev_txns should be.
         # Since we know that we are only sending one transaction at a time,
         # we can simply take the last one.
@@ -143,7 +145,7 @@ class TransactionStore(SQLBaseStore):
             txn,
             table=SentTransactions.table_name,
             values={
-                "id": self.get_next_stream_id(),
+                "id": next_id,
                 "transaction_id": transaction_id,
                 "destination": destination,
                 "ts": origin_server_ts,
-- 
cgit 1.4.1


From 0af5f5efaf0b24187514cf78d7982ef9b85a208c Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Tue, 7 Apr 2015 12:08:35 +0100
Subject: Don't use multiple UNIQUE constraints; it will cause deadlocks

---
 synapse/storage/_base.py                            |  4 ++--
 synapse/storage/events.py                           | 14 +++++++-------
 synapse/storage/schema/full_schemas/11/presence.sql | 10 ++++++----
 synapse/storage/schema/full_schemas/11/users.sql    |  4 ++--
 4 files changed, 17 insertions(+), 15 deletions(-)

(limited to 'synapse/storage/events.py')

diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index a0c1718c27..4ac61be895 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -837,11 +837,11 @@ class SQLBaseStore(object):
             return curr_time
 
         logger.debug("Got js: %r", js)
-        d = json.loads(js)
+        d = json.loads(str(js).decode("utf8"))
         start_time = update_counter("decode_json", start_time)
 
         logger.debug("Got internal_metadata: %r", internal_metadata)
-        internal_metadata = json.loads(internal_metadata)
+        internal_metadata = json.loads(str(internal_metadata).decode("utf8"))
         start_time = update_counter("decode_internal", start_time)
 
         ev = FrozenEvent(d, internal_metadata_dict=internal_metadata)
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 514feebcbf..3b3416716e 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -179,7 +179,7 @@ class EventsStore(SQLBaseStore):
                 )
                 txn.execute(
                     sql,
-                    (metadata_json.decode("UTF-8"), event.event_id,)
+                    (buffer(metadata_json), event.event_id,)
                 )
 
                 sql = (
@@ -224,14 +224,14 @@ class EventsStore(SQLBaseStore):
             values={
                 "event_id": event.event_id,
                 "room_id": event.room_id,
-                "internal_metadata": metadata_json.decode("UTF-8"),
-                "json": encode_canonical_json(event_dict).decode("UTF-8"),
+                "internal_metadata": buffer(metadata_json),
+                "json": buffer(encode_canonical_json(event_dict)),
             },
         )
 
-        content = encode_canonical_json(
+        content = buffer(encode_canonical_json(
             event.content
-        ).decode("UTF-8")
+        ))
 
         vals = {
             "topological_ordering": event.depth,
@@ -256,9 +256,9 @@ class EventsStore(SQLBaseStore):
             ]
         }
 
-        vals["unrecognized_keys"] = encode_canonical_json(
+        vals["unrecognized_keys"] = buffer(encode_canonical_json(
             unrec
-        ).decode("UTF-8")
+        ))
 
         sql = (
             "INSERT INTO events"
diff --git a/synapse/storage/schema/full_schemas/11/presence.sql b/synapse/storage/schema/full_schemas/11/presence.sql
index 273e61281a..00d803a5cd 100644
--- a/synapse/storage/schema/full_schemas/11/presence.sql
+++ b/synapse/storage/schema/full_schemas/11/presence.sql
@@ -17,20 +17,22 @@ CREATE TABLE IF NOT EXISTS presence(
   state VARCHAR(20),
   status_msg VARCHAR(150),
   mtime BIGINT, -- miliseconds since last state change
-  UNIQUE(user_id)
+  UNIQUE (user_id)
 ) ;
 
 -- For each of /my/ users which possibly-remote users are allowed to see their
 -- presence state
 CREATE TABLE IF NOT EXISTS presence_allow_inbound(
   observed_user_id VARCHAR(150) NOT NULL,
-  observer_user_id VARCHAR(150) -- a UserID,
+  observer_user_id VARCHAR(150) NOT NULL, -- a UserID,
+  UNIQUE (observed_user_id, observer_user_id)
 ) ;
 
 -- For each of /my/ users (watcher), which possibly-remote users are they
 -- watching?
 CREATE TABLE IF NOT EXISTS presence_list(
   user_id VARCHAR(150) NOT NULL,
-  observed_user_id VARCHAR(150), -- a UserID,
-  accepted BOOLEAN
+  observed_user_id VARCHAR(150) NOT NULL, -- a UserID,
+  accepted BOOLEAN NOT NULL,
+  UNIQUE (user_id, observed_user_id)
 ) ;
diff --git a/synapse/storage/schema/full_schemas/11/users.sql b/synapse/storage/schema/full_schemas/11/users.sql
index 0271de3526..ba0f42d455 100644
--- a/synapse/storage/schema/full_schemas/11/users.sql
+++ b/synapse/storage/schema/full_schemas/11/users.sql
@@ -35,8 +35,8 @@ CREATE TABLE IF NOT EXISTS user_ips (
     device_id VARCHAR(150),
     ip VARCHAR(150) NOT NULL,
     user_agent VARCHAR(150) NOT NULL,
-    last_seen BIGINT NOT NULL,
-    UNIQUE (user, access_token, ip, user_agent)
+    last_seen BIGINT NOT NULL
 ) ;
 
 CREATE INDEX IF NOT EXISTS user_ips_user ON user_ips(user);
+CREATE INDEX IF NOT EXISTS user_ips_user_ip ON user_ips(user, access_token, ip);
-- 
cgit 1.4.1


From 8ad0f4912ed72daced74ae4d1c939ebdbc517476 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 9 Apr 2015 11:41:36 +0100
Subject: Stream ordering and out of order insertions.

Handle the fact that events can be persisted out of order, and so to get
the "current max" stream token becomes non trivial - as we need to make
sure that *all* stream tokens less than the current max have also
successfully been persisted.
---
 synapse/storage/_base.py              |  46 +------------
 synapse/storage/events.py             |   9 ++-
 synapse/storage/stream.py             |  23 +------
 synapse/storage/util/__init__.py      |  14 ++++
 synapse/storage/util/id_generators.py | 126 ++++++++++++++++++++++++++++++++++
 5 files changed, 153 insertions(+), 65 deletions(-)
 create mode 100644 synapse/storage/util/__init__.py
 create mode 100644 synapse/storage/util/id_generators.py

(limited to 'synapse/storage/events.py')

diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 23289bbdd4..badf9a5f40 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -22,6 +22,8 @@ from synapse.util.logcontext import PreserveLoggingContext, LoggingContext
 from synapse.util.lrucache import LruCache
 import synapse.metrics
 
+from util.id_generators import IdGenerator, StreamIdGenerator
+
 from twisted.internet import defer
 
 from collections import namedtuple, OrderedDict
@@ -29,7 +31,6 @@ import functools
 import simplejson as json
 import sys
 import time
-import threading
 
 
 logger = logging.getLogger(__name__)
@@ -232,46 +233,6 @@ class PerformanceCounters(object):
         return top_n_counters
 
 
-class IdGenerator(object):
-    def __init__(self, table, column, store):
-        self.table = table
-        self.column = column
-        self.store = store
-        self._lock = threading.Lock()
-        self._next_id = None
-
-    @defer.inlineCallbacks
-    def get_next(self):
-        with self._lock:
-            if not self._next_id:
-                res = yield self.store._execute_and_decode(
-                    "IdGenerator_%s" % (self.table,),
-                    "SELECT MAX(%s) as mx FROM %s" % (self.column, self.table,)
-                )
-
-                self._next_id = (res and res[0] and res[0]["mx"]) or 1
-
-            i = self._next_id
-            self._next_id += 1
-            defer.returnValue(i)
-
-    def get_next_txn(self, txn):
-        with self._lock:
-            if self._next_id:
-                i = self._next_id
-                self._next_id += 1
-                return i
-            else:
-                txn.execute(
-                    "SELECT MAX(%s) FROM %s" % (self.column, self.table,)
-                )
-
-                val, = txn.fetchone()
-                self._next_id = val or 2
-
-                return 1
-
-
 class SQLBaseStore(object):
     _TXN_ID = 0
 
@@ -297,7 +258,7 @@ class SQLBaseStore(object):
         # Pretend the getEventCache is just another named cache
         caches_by_name["*getEvent*"] = self._get_event_cache
 
-        self._stream_id_gen = IdGenerator("events", "stream_ordering", self)
+        self._stream_id_gen = StreamIdGenerator()
         self._transaction_id_gen = IdGenerator("sent_transactions", "id", self)
         self._state_groups_id_gen = IdGenerator("state_groups", "id", self)
         self._access_tokens_id_gen = IdGenerator("access_tokens", "id", self)
@@ -363,7 +324,6 @@ class SQLBaseStore(object):
                                 *args, **kwargs
                             )
                         except self.database_engine.module.DatabaseError as e:
-                            logger.warn("[TXN DEADLOCK] {%s} %r, %r", name, e.errno, e.sqlstate)
                             if self.database_engine.is_deadlock(e):
                                 logger.warn("[TXN DEADLOCK] {%s} %d/%d", name, i, N)
                                 if i < N:
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 3b3416716e..f066484c7e 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -52,7 +52,6 @@ class EventsStore(SQLBaseStore):
                 is_new_state=is_new_state,
                 current_state=current_state,
             )
-            self.get_room_events_max_id.invalidate()
         except _RollbackButIsFineException:
             pass
 
@@ -97,7 +96,13 @@ class EventsStore(SQLBaseStore):
         self._get_event_cache.pop(event.event_id)
 
         if stream_ordering is None:
-            stream_ordering = self._stream_id_gen.get_next_txn(txn)
+            with self._stream_id_gen.get_next_txn(txn) as stream_ordering:
+                return self._persist_event_txn(
+                    txn, event, context, backfilled,
+                    stream_ordering=stream_ordering,
+                    is_new_state=is_new_state,
+                    current_state=current_state,
+                )
 
         # We purposefully do this first since if we include a `current_state`
         # key, we *want* to update the `current_state_events` table
diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py
index e6bb5a8077..9925f04bf7 100644
--- a/synapse/storage/stream.py
+++ b/synapse/storage/stream.py
@@ -413,12 +413,10 @@ class StreamStore(SQLBaseStore):
             "get_recent_events_for_room", get_recent_events_for_room_txn
         )
 
-    @cached(num_args=0)
+    @defer.inlineCallbacks
     def get_room_events_max_id(self):
-        return self.runInteraction(
-            "get_room_events_max_id",
-            self._get_room_events_max_id_txn
-        )
+        token = yield self._stream_id_gen.get_max_token(self)
+        defer.returnValue("s%d" % (token,))
 
     @defer.inlineCallbacks
     def _get_min_token(self):
@@ -433,21 +431,6 @@ class StreamStore(SQLBaseStore):
 
         defer.returnValue(self.min_token)
 
-    def _get_room_events_max_id_txn(self, txn):
-        txn.execute(
-            "SELECT MAX(stream_ordering) as m FROM events"
-        )
-
-        res = self.cursor_to_dict(txn)
-
-        logger.debug("get_room_events_max_id: %s", res)
-
-        if not res or not res[0] or not res[0]["m"]:
-            return "s0"
-
-        key = res[0]["m"]
-        return "s%d" % (key,)
-
     @staticmethod
     def _set_before_and_after(events, rows):
         for event, row in zip(events, rows):
diff --git a/synapse/storage/util/__init__.py b/synapse/storage/util/__init__.py
new file mode 100644
index 0000000000..c488b10d3c
--- /dev/null
+++ b/synapse/storage/util/__init__.py
@@ -0,0 +1,14 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014, 2015 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.
diff --git a/synapse/storage/util/id_generators.py b/synapse/storage/util/id_generators.py
new file mode 100644
index 0000000000..8f419323a7
--- /dev/null
+++ b/synapse/storage/util/id_generators.py
@@ -0,0 +1,126 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014, 2015 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.
+
+from twisted.internet import defer
+
+from collections import deque
+import contextlib
+import threading
+
+
+class IdGenerator(object):
+    def __init__(self, table, column, store):
+        self.table = table
+        self.column = column
+        self.store = store
+        self._lock = threading.Lock()
+        self._next_id = None
+
+    @defer.inlineCallbacks
+    def get_next(self):
+        with self._lock:
+            if not self._next_id:
+                res = yield self.store._execute_and_decode(
+                    "IdGenerator_%s" % (self.table,),
+                    "SELECT MAX(%s) as mx FROM %s" % (self.column, self.table,)
+                )
+
+                self._next_id = (res and res[0] and res[0]["mx"]) or 1
+
+            i = self._next_id
+            self._next_id += 1
+            defer.returnValue(i)
+
+    def get_next_txn(self, txn):
+        with self._lock:
+            if self._next_id:
+                i = self._next_id
+                self._next_id += 1
+                return i
+            else:
+                txn.execute(
+                    "SELECT MAX(%s) FROM %s" % (self.column, self.table,)
+                )
+
+                val, = txn.fetchone()
+                self._next_id = val or 2
+
+                return 1
+
+
+class StreamIdGenerator(object):
+    """Used to generate new stream ids when persisting events while keeping
+    track of which transactions have been completed.
+
+    This allows us to get the "current" stream id, i.e. the stream id such that
+    all ids less than or equal to it have completed. This handles the fact that
+    persistence of events can complete out of order.
+
+    Usage:
+        with stream_id_gen.get_next_txn(txn) as stream_id:
+            # ... persist event ...
+    """
+    def __init__(self):
+        self._lock = threading.Lock()
+
+        self._current_max = None
+        self._unfinished_ids = deque()
+
+    def get_next_txn(self, txn):
+        """
+        Usage:
+            with stream_id_gen.get_next_txn(txn) as stream_id:
+                # ... persist event ...
+        """
+        with self._lock:
+            if not self._current_max:
+                self._compute_current_max(txn)
+
+            self._current_max += 1
+            next_id = self._current_max
+
+            self._unfinished_ids.append(next_id)
+
+        @contextlib.contextmanager
+        def manager():
+            yield next_id
+            with self._lock:
+                self._unfinished_ids.remove(next_id)
+
+        return manager()
+
+    def get_max_token(self, store):
+        """Returns the maximum stream id such that all stream ids less than or
+        equal to it have been successfully persisted.
+        """
+        with self._lock:
+            if self._unfinished_ids:
+                return self._unfinished_ids[0] - 1
+
+            if not self._current_max:
+                return store.runInteraction(
+                    "_compute_current_max",
+                    self._compute_current_max,
+                )
+
+            return self._current_max
+
+    def _compute_current_max(self, txn):
+        txn.execute("SELECT MAX(stream_ordering) FROM events")
+        val, = txn.fetchone()
+
+        self._current_max = int(val) if val else 1
+
+        return self._current_max
-- 
cgit 1.4.1


From 58d83399663a080c123d2f112b4f4d84accbc638 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Tue, 14 Apr 2015 13:53:20 +0100
Subject: Add support for postgres instead of mysql. Change sql accourdingly.
 blob + varbinary -> bytea. No support for UNSIGNED or CREATE INDEX IF NOT
 EXISTS.

---
 synapse/app/homeserver.py                          |  2 +
 synapse/storage/__init__.py                        | 15 +++-----
 synapse/storage/_base.py                           |  2 +-
 synapse/storage/engines/__init__.py                |  2 +
 synapse/storage/event_federation.py                | 10 ++---
 synapse/storage/events.py                          |  4 +-
 synapse/storage/room.py                            | 34 ++++++++++++-----
 .../full_schemas/16/application_services.sql       | 10 ++---
 .../storage/schema/full_schemas/16/event_edges.sql | 26 ++++++-------
 .../schema/full_schemas/16/event_signatures.sql    | 16 ++++----
 synapse/storage/schema/full_schemas/16/im.sql      | 44 +++++++++++-----------
 synapse/storage/schema/full_schemas/16/keys.sql    |  8 ++--
 .../schema/full_schemas/16/media_repository.sql    |  8 ++--
 .../storage/schema/full_schemas/16/presence.sql    |  4 +-
 synapse/storage/schema/full_schemas/16/push.sql    | 30 +++++++--------
 .../storage/schema/full_schemas/16/redactions.sql  |  4 +-
 synapse/storage/schema/full_schemas/16/state.sql   | 10 ++---
 .../schema/full_schemas/16/transactions.sql        | 24 ++++++------
 synapse/storage/schema/full_schemas/16/users.sql   | 10 ++---
 synapse/storage/schema/schema_version.sql          | 14 +++----
 synapse/storage/stream.py                          | 16 ++++----
 21 files changed, 153 insertions(+), 140 deletions(-)

(limited to 'synapse/storage/events.py')

diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py
index a47e548d66..033011e1d7 100755
--- a/synapse/app/homeserver.py
+++ b/synapse/app/homeserver.py
@@ -373,6 +373,8 @@ def setup(config_options):
             "use_unicode": True,
             "collation": "utf8mb4_bin",
         })
+    elif name == "psycopg2":
+        pass
     elif name == "sqlite3":
         db_config.setdefault("args", {}).update({
             "cp_min": 1,
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index b46cafd25e..272420194d 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -236,7 +236,7 @@ def _setup_new_database(cur, database_engine):
 
     cur.execute(
         database_engine.convert_param_style(
-            "REPLACE INTO schema_version (version, upgraded)"
+            "INSERT INTO schema_version (version, upgraded)"
             " VALUES (?,?)"
         ),
         (max_current_ver, False,)
@@ -432,14 +432,11 @@ def executescript(txn, schema_path):
 
 
 def _get_or_create_schema_state(txn, database_engine):
-    try:
-        # Bluntly try creating the schema_version tables.
-        schema_path = os.path.join(
-            dir_path, "schema", "schema_version.sql",
-        )
-        executescript(txn, schema_path)
-    except:
-        pass
+    # Bluntly try creating the schema_version tables.
+    schema_path = os.path.join(
+        dir_path, "schema", "schema_version.sql",
+    )
+    executescript(txn, schema_path)
 
     txn.execute("SELECT version, upgraded FROM schema_version")
     row = txn.fetchone()
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index e30514cd5e..fa5199104a 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -330,7 +330,7 @@ class SQLBaseStore(object):
                                     continue
                             raise
                 except Exception as e:
-                    logger.debug("[TXN FAIL] {%s}", name, e)
+                    logger.debug("[TXN FAIL] {%s} %s", name, e)
                     raise
                 finally:
                     end = time.time() * 1000
diff --git a/synapse/storage/engines/__init__.py b/synapse/storage/engines/__init__.py
index 29702be923..548d4e1b42 100644
--- a/synapse/storage/engines/__init__.py
+++ b/synapse/storage/engines/__init__.py
@@ -14,6 +14,7 @@
 # limitations under the License.
 
 from .maria import MariaEngine
+from .postgres import PostgresEngine
 from .sqlite3 import Sqlite3Engine
 
 import importlib
@@ -22,6 +23,7 @@ import importlib
 SUPPORTED_MODULE = {
     "sqlite3": Sqlite3Engine,
     "mysql.connector": MariaEngine,
+    "psycopg2": PostgresEngine,
 }
 
 
diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py
index 79ad5ddc9c..54a3c9d805 100644
--- a/synapse/storage/event_federation.py
+++ b/synapse/storage/event_federation.py
@@ -153,7 +153,7 @@ class EventFederationStore(SQLBaseStore):
         results = self._get_prev_events_and_state(
             txn,
             event_id,
-            is_state=1,
+            is_state=True,
         )
 
         return [(e_id, h, ) for e_id, h, _ in results]
@@ -164,7 +164,7 @@ class EventFederationStore(SQLBaseStore):
         }
 
         if is_state is not None:
-            keyvalues["is_state"] = is_state
+            keyvalues["is_state"] = bool(is_state)
 
         res = self._simple_select_list_txn(
             txn,
@@ -259,7 +259,7 @@ class EventFederationStore(SQLBaseStore):
                     "event_id": event_id,
                     "prev_event_id": e_id,
                     "room_id": room_id,
-                    "is_state": 0,
+                    "is_state": False,
                 },
             )
 
@@ -397,7 +397,7 @@ class EventFederationStore(SQLBaseStore):
 
         query = (
             "SELECT prev_event_id FROM event_edges "
-            "WHERE room_id = ? AND event_id = ? AND is_state = 0 "
+            "WHERE room_id = ? AND event_id = ? AND is_state = ? "
             "LIMIT ?"
         )
 
@@ -406,7 +406,7 @@ class EventFederationStore(SQLBaseStore):
             for event_id in front:
                 txn.execute(
                     query,
-                    (room_id, event_id, limit - len(event_results))
+                    (room_id, event_id, False, limit - len(event_results))
                 )
 
                 for e_id, in txn.fetchall():
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index a2e87c27ce..9fe2effb4b 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -188,12 +188,12 @@ class EventsStore(SQLBaseStore):
                 )
 
                 sql = (
-                    "UPDATE events SET outlier = 0"
+                    "UPDATE events SET outlier = ?"
                     " WHERE event_id = ?"
                 )
                 txn.execute(
                     sql,
-                    (event.event_id,)
+                    (False, event.event_id,)
                 )
             return
 
diff --git a/synapse/storage/room.py b/synapse/storage/room.py
index a1a76280fe..48ebb33057 100644
--- a/synapse/storage/room.py
+++ b/synapse/storage/room.py
@@ -105,14 +105,12 @@ class RoomStore(SQLBaseStore):
 
             # We use non printing ascii character US (\x1F) as a separator
             sql = (
-                "SELECT r.room_id, n.name, t.topic, "
-                "group_concat(a.room_alias, '\x1F') "
-                "FROM rooms AS r "
-                "LEFT JOIN (%(topic)s) AS t ON t.room_id = r.room_id "
-                "LEFT JOIN (%(name)s) AS n ON n.room_id = r.room_id "
-                "INNER JOIN room_aliases AS a ON a.room_id = r.room_id "
-                "WHERE r.is_public = ? "
-                "GROUP BY r.room_id "
+                "SELECT r.room_id, max(n.name), max(t.topic)"
+                " FROM rooms AS r"
+                " LEFT JOIN (%(topic)s) AS t ON t.room_id = r.room_id"
+                " LEFT JOIN (%(name)s) AS n ON n.room_id = r.room_id"
+                " WHERE r.is_public = ?"
+                " GROUP BY r.room_id"
             ) % {
                 "topic": topic_subquery,
                 "name": name_subquery,
@@ -120,7 +118,22 @@ class RoomStore(SQLBaseStore):
 
             txn.execute(sql, (is_public,))
 
-            return txn.fetchall()
+            rows = txn.fetchall()
+
+            for i, row in enumerate(rows):
+                room_id = row[0]
+                aliases = self._simple_select_onecol_txn(
+                    txn,
+                    table="room_aliases",
+                    keyvalues={
+                        "room_id": room_id
+                    },
+                    retcol="room_alias",
+                )
+
+                rows[i] = list(row) + [aliases]
+
+            return rows
 
         rows = yield self.runInteraction(
             "get_rooms", f
@@ -131,9 +144,10 @@ class RoomStore(SQLBaseStore):
                 "room_id": r[0],
                 "name": r[1],
                 "topic": r[2],
-                "aliases": r[3].split("\x1F"),
+                "aliases": r[3],
             }
             for r in rows
+            if r[3]  # We only return rooms that have at least one alias.
         ]
 
         defer.returnValue(ret)
diff --git a/synapse/storage/schema/full_schemas/16/application_services.sql b/synapse/storage/schema/full_schemas/16/application_services.sql
index bc709df92d..f08c5bcf76 100644
--- a/synapse/storage/schema/full_schemas/16/application_services.sql
+++ b/synapse/storage/schema/full_schemas/16/application_services.sql
@@ -14,7 +14,7 @@
  */
 
 CREATE TABLE IF NOT EXISTS application_services(
-    id BIGINT UNSIGNED PRIMARY KEY,
+    id BIGINT PRIMARY KEY,
     url VARCHAR(150),
     token VARCHAR(150),
     hs_token VARCHAR(150),
@@ -23,8 +23,8 @@ CREATE TABLE IF NOT EXISTS application_services(
 );
 
 CREATE TABLE IF NOT EXISTS application_services_regex(
-    id BIGINT UNSIGNED PRIMARY KEY,
-    as_id BIGINT UNSIGNED NOT NULL,
+    id BIGINT PRIMARY KEY,
+    as_id BIGINT NOT NULL,
     namespace INTEGER,  /* enum[room_id|room_alias|user_id] */
     regex VARCHAR(150),
     FOREIGN KEY(as_id) REFERENCES application_services(id)
@@ -39,10 +39,10 @@ CREATE TABLE IF NOT EXISTS application_services_state(
 CREATE TABLE IF NOT EXISTS application_services_txns(
     as_id VARCHAR(150) NOT NULL,
     txn_id INTEGER NOT NULL,
-    event_ids LONGBLOB NOT NULL,
+    event_ids bytea NOT NULL,
     UNIQUE(as_id, txn_id)
 );
 
-CREATE INDEX IF NOT EXISTS application_services_txns_id ON application_services_txns (
+CREATE INDEX application_services_txns_id ON application_services_txns (
     as_id
 );
diff --git a/synapse/storage/schema/full_schemas/16/event_edges.sql b/synapse/storage/schema/full_schemas/16/event_edges.sql
index bdb1109094..05d0874f0d 100644
--- a/synapse/storage/schema/full_schemas/16/event_edges.sql
+++ b/synapse/storage/schema/full_schemas/16/event_edges.sql
@@ -19,8 +19,8 @@ CREATE TABLE IF NOT EXISTS event_forward_extremities(
     UNIQUE (event_id, room_id)
 );
 
-CREATE INDEX IF NOT EXISTS ev_extrem_room ON event_forward_extremities(room_id);
-CREATE INDEX IF NOT EXISTS ev_extrem_id ON event_forward_extremities(event_id);
+CREATE INDEX ev_extrem_room ON event_forward_extremities(room_id);
+CREATE INDEX ev_extrem_id ON event_forward_extremities(event_id);
 
 
 CREATE TABLE IF NOT EXISTS event_backward_extremities(
@@ -29,8 +29,8 @@ CREATE TABLE IF NOT EXISTS event_backward_extremities(
     UNIQUE (event_id, room_id)
 );
 
-CREATE INDEX IF NOT EXISTS ev_b_extrem_room ON event_backward_extremities(room_id);
-CREATE INDEX IF NOT EXISTS ev_b_extrem_id ON event_backward_extremities(event_id);
+CREATE INDEX ev_b_extrem_room ON event_backward_extremities(room_id);
+CREATE INDEX ev_b_extrem_id ON event_backward_extremities(event_id);
 
 
 CREATE TABLE IF NOT EXISTS event_edges(
@@ -41,8 +41,8 @@ CREATE TABLE IF NOT EXISTS event_edges(
     UNIQUE (event_id, prev_event_id, room_id, is_state)
 );
 
-CREATE INDEX IF NOT EXISTS ev_edges_id ON event_edges(event_id);
-CREATE INDEX IF NOT EXISTS ev_edges_prev_id ON event_edges(prev_event_id);
+CREATE INDEX ev_edges_id ON event_edges(event_id);
+CREATE INDEX ev_edges_prev_id ON event_edges(prev_event_id);
 
 
 CREATE TABLE IF NOT EXISTS room_depth(
@@ -51,17 +51,17 @@ CREATE TABLE IF NOT EXISTS room_depth(
     UNIQUE (room_id)
 );
 
-CREATE INDEX IF NOT EXISTS room_depth_room ON room_depth(room_id);
+CREATE INDEX room_depth_room ON room_depth(room_id);
 
 
 create TABLE IF NOT EXISTS event_destinations(
     event_id VARCHAR(150) NOT NULL,
     destination VARCHAR(150) NOT NULL,
-    delivered_ts BIGINT UNSIGNED DEFAULT 0, -- or 0 if not delivered
+    delivered_ts BIGINT DEFAULT 0, -- or 0 if not delivered
     UNIQUE (event_id, destination)
 );
 
-CREATE INDEX IF NOT EXISTS event_destinations_id ON event_destinations(event_id);
+CREATE INDEX event_destinations_id ON event_destinations(event_id);
 
 
 CREATE TABLE IF NOT EXISTS state_forward_extremities(
@@ -72,10 +72,10 @@ CREATE TABLE IF NOT EXISTS state_forward_extremities(
     UNIQUE (event_id, room_id)
 );
 
-CREATE INDEX IF NOT EXISTS st_extrem_keys ON state_forward_extremities(
+CREATE INDEX st_extrem_keys ON state_forward_extremities(
     room_id, type, state_key
 );
-CREATE INDEX IF NOT EXISTS st_extrem_id ON state_forward_extremities(event_id);
+CREATE INDEX st_extrem_id ON state_forward_extremities(event_id);
 
 
 CREATE TABLE IF NOT EXISTS event_auth(
@@ -85,5 +85,5 @@ CREATE TABLE IF NOT EXISTS event_auth(
     UNIQUE (event_id, auth_id, room_id)
 );
 
-CREATE INDEX IF NOT EXISTS evauth_edges_id ON event_auth(event_id);
-CREATE INDEX IF NOT EXISTS evauth_edges_auth_id ON event_auth(auth_id);
+CREATE INDEX evauth_edges_id ON event_auth(event_id);
+CREATE INDEX evauth_edges_auth_id ON event_auth(auth_id);
diff --git a/synapse/storage/schema/full_schemas/16/event_signatures.sql b/synapse/storage/schema/full_schemas/16/event_signatures.sql
index 09886f607c..4291827368 100644
--- a/synapse/storage/schema/full_schemas/16/event_signatures.sql
+++ b/synapse/storage/schema/full_schemas/16/event_signatures.sql
@@ -16,40 +16,40 @@
 CREATE TABLE IF NOT EXISTS event_content_hashes (
     event_id VARCHAR(150),
     algorithm VARCHAR(150),
-    hash LONGBLOB,
+    hash bytea,
     UNIQUE (event_id, algorithm)
 );
 
-CREATE INDEX IF NOT EXISTS event_content_hashes_id ON event_content_hashes(event_id);
+CREATE INDEX event_content_hashes_id ON event_content_hashes(event_id);
 
 
 CREATE TABLE IF NOT EXISTS event_reference_hashes (
     event_id VARCHAR(150),
     algorithm VARCHAR(150),
-    hash LONGBLOB,
+    hash bytea,
     UNIQUE (event_id, algorithm)
 );
 
-CREATE INDEX IF NOT EXISTS event_reference_hashes_id ON event_reference_hashes(event_id);
+CREATE INDEX event_reference_hashes_id ON event_reference_hashes(event_id);
 
 
 CREATE TABLE IF NOT EXISTS event_signatures (
     event_id VARCHAR(150),
     signature_name VARCHAR(150),
     key_id VARCHAR(150),
-    signature LONGBLOB,
+    signature bytea,
     UNIQUE (event_id, signature_name, key_id)
 );
 
-CREATE INDEX IF NOT EXISTS event_signatures_id ON event_signatures(event_id);
+CREATE INDEX event_signatures_id ON event_signatures(event_id);
 
 
 CREATE TABLE IF NOT EXISTS event_edge_hashes(
     event_id VARCHAR(150),
     prev_event_id VARCHAR(150),
     algorithm VARCHAR(150),
-    hash LONGBLOB,
+    hash bytea,
     UNIQUE (event_id, prev_event_id, algorithm)
 );
 
-CREATE INDEX IF NOT EXISTS event_edge_hashes_id ON event_edge_hashes(event_id);
+CREATE INDEX event_edge_hashes_id ON event_edge_hashes(event_id);
diff --git a/synapse/storage/schema/full_schemas/16/im.sql b/synapse/storage/schema/full_schemas/16/im.sql
index 19f0f34143..a661fc160c 100644
--- a/synapse/storage/schema/full_schemas/16/im.sql
+++ b/synapse/storage/schema/full_schemas/16/im.sql
@@ -14,33 +14,33 @@
  */
 
 CREATE TABLE IF NOT EXISTS events(
-    stream_ordering BIGINT UNSIGNED PRIMARY KEY,
-    topological_ordering BIGINT UNSIGNED NOT NULL,
+    stream_ordering BIGINT PRIMARY KEY,
+    topological_ordering BIGINT NOT NULL,
     event_id VARCHAR(150) NOT NULL,
     type VARCHAR(150) NOT NULL,
     room_id VARCHAR(150) NOT NULL,
-    content LONGBLOB NOT NULL,
-    unrecognized_keys LONGBLOB,
+    content bytea NOT NULL,
+    unrecognized_keys bytea,
     processed BOOL NOT NULL,
     outlier BOOL NOT NULL,
-    depth BIGINT UNSIGNED DEFAULT 0 NOT NULL,
+    depth BIGINT DEFAULT 0 NOT NULL,
     UNIQUE (event_id)
 );
 
-CREATE INDEX IF NOT EXISTS events_stream_ordering ON events (stream_ordering);
-CREATE INDEX IF NOT EXISTS events_topological_ordering ON events (topological_ordering);
-CREATE INDEX IF NOT EXISTS events_room_id ON events (room_id);
+CREATE INDEX events_stream_ordering ON events (stream_ordering);
+CREATE INDEX events_topological_ordering ON events (topological_ordering);
+CREATE INDEX events_room_id ON events (room_id);
 
 
 CREATE TABLE IF NOT EXISTS event_json(
     event_id VARCHAR(150) NOT NULL,
     room_id VARCHAR(150) NOT NULL,
-    internal_metadata LONGBLOB NOT NULL,
-    json LONGBLOB NOT NULL,
+    internal_metadata bytea NOT NULL,
+    json bytea NOT NULL,
     UNIQUE (event_id)
 );
 
-CREATE INDEX IF NOT EXISTS event_json_room_id ON event_json(room_id);
+CREATE INDEX event_json_room_id ON event_json(room_id);
 
 
 CREATE TABLE IF NOT EXISTS state_events(
@@ -52,9 +52,9 @@ CREATE TABLE IF NOT EXISTS state_events(
     UNIQUE (event_id)
 );
 
-CREATE INDEX IF NOT EXISTS state_events_room_id ON state_events (room_id);
-CREATE INDEX IF NOT EXISTS state_events_type ON state_events (type);
-CREATE INDEX IF NOT EXISTS state_events_state_key ON state_events (state_key);
+CREATE INDEX state_events_room_id ON state_events (room_id);
+CREATE INDEX state_events_type ON state_events (type);
+CREATE INDEX state_events_state_key ON state_events (state_key);
 
 
 CREATE TABLE IF NOT EXISTS current_state_events(
@@ -66,9 +66,9 @@ CREATE TABLE IF NOT EXISTS current_state_events(
     UNIQUE (room_id, type, state_key)
 );
 
-CREATE INDEX IF NOT EXISTS current_state_events_room_id ON current_state_events (room_id);
-CREATE INDEX IF NOT EXISTS current_state_events_type ON current_state_events (type);
-CREATE INDEX IF NOT EXISTS current_state_events_state_key ON current_state_events (state_key);
+CREATE INDEX current_state_events_room_id ON current_state_events (room_id);
+CREATE INDEX current_state_events_type ON current_state_events (type);
+CREATE INDEX current_state_events_state_key ON current_state_events (state_key);
 
 CREATE TABLE IF NOT EXISTS room_memberships(
     event_id VARCHAR(150) NOT NULL,
@@ -79,8 +79,8 @@ CREATE TABLE IF NOT EXISTS room_memberships(
     UNIQUE (event_id)
 );
 
-CREATE INDEX IF NOT EXISTS room_memberships_room_id ON room_memberships (room_id);
-CREATE INDEX IF NOT EXISTS room_memberships_user_id ON room_memberships (user_id);
+CREATE INDEX room_memberships_room_id ON room_memberships (room_id);
+CREATE INDEX room_memberships_user_id ON room_memberships (user_id);
 
 CREATE TABLE IF NOT EXISTS feedback(
     event_id VARCHAR(150) NOT NULL,
@@ -98,7 +98,7 @@ CREATE TABLE IF NOT EXISTS topics(
     UNIQUE (event_id)
 );
 
-CREATE INDEX IF NOT EXISTS topics_room_id ON topics(room_id);
+CREATE INDEX topics_room_id ON topics(room_id);
 
 CREATE TABLE IF NOT EXISTS room_names(
     event_id VARCHAR(150) NOT NULL,
@@ -107,7 +107,7 @@ CREATE TABLE IF NOT EXISTS room_names(
     UNIQUE (event_id)
 );
 
-CREATE INDEX IF NOT EXISTS room_names_room_id ON room_names(room_id);
+CREATE INDEX room_names_room_id ON room_names(room_id);
 
 CREATE TABLE IF NOT EXISTS rooms(
     room_id VARCHAR(150) PRIMARY KEY NOT NULL,
@@ -121,4 +121,4 @@ CREATE TABLE IF NOT EXISTS room_hosts(
     UNIQUE (room_id, host)
 );
 
-CREATE INDEX IF NOT EXISTS room_hosts_room_id ON room_hosts (room_id);
+CREATE INDEX room_hosts_room_id ON room_hosts (room_id);
diff --git a/synapse/storage/schema/full_schemas/16/keys.sql b/synapse/storage/schema/full_schemas/16/keys.sql
index 35f141c288..459b510427 100644
--- a/synapse/storage/schema/full_schemas/16/keys.sql
+++ b/synapse/storage/schema/full_schemas/16/keys.sql
@@ -16,8 +16,8 @@ CREATE TABLE IF NOT EXISTS server_tls_certificates(
   server_name VARCHAR(150), -- Server name.
   fingerprint VARCHAR(150), -- Certificate fingerprint.
   from_server VARCHAR(150), -- Which key server the certificate was fetched from.
-  ts_added_ms BIGINT UNSIGNED, -- When the certifcate was added.
-  tls_certificate LONGBLOB, -- DER encoded x509 certificate.
+  ts_added_ms BIGINT, -- When the certifcate was added.
+  tls_certificate bytea, -- DER encoded x509 certificate.
   UNIQUE (server_name, fingerprint)
 );
 
@@ -25,7 +25,7 @@ CREATE TABLE IF NOT EXISTS server_signature_keys(
   server_name VARCHAR(150), -- Server name.
   key_id VARCHAR(150), -- Key version.
   from_server VARCHAR(150), -- Which key server the key was fetched form.
-  ts_added_ms BIGINT UNSIGNED, -- When the key was added.
-  verify_key LONGBLOB, -- NACL verification key.
+  ts_added_ms BIGINT, -- When the key was added.
+  verify_key bytea, -- NACL verification key.
   UNIQUE (server_name, key_id)
 );
diff --git a/synapse/storage/schema/full_schemas/16/media_repository.sql b/synapse/storage/schema/full_schemas/16/media_repository.sql
index 014bce4aeb..0e819fca38 100644
--- a/synapse/storage/schema/full_schemas/16/media_repository.sql
+++ b/synapse/storage/schema/full_schemas/16/media_repository.sql
@@ -17,7 +17,7 @@ CREATE TABLE IF NOT EXISTS local_media_repository (
     media_id VARCHAR(150), -- The id used to refer to the media.
     media_type VARCHAR(150), -- The MIME-type of the media.
     media_length INTEGER, -- Length of the media in bytes.
-    created_ts BIGINT UNSIGNED, -- When the content was uploaded in ms.
+    created_ts BIGINT, -- When the content was uploaded in ms.
     upload_name VARCHAR(150), -- The name the media was uploaded with.
     user_id VARCHAR(150), -- The user who uploaded the file.
     UNIQUE (media_id)
@@ -35,14 +35,14 @@ CREATE TABLE IF NOT EXISTS local_media_repository_thumbnails (
     )
 );
 
-CREATE INDEX IF NOT EXISTS local_media_repository_thumbnails_media_id
+CREATE INDEX local_media_repository_thumbnails_media_id
     ON local_media_repository_thumbnails (media_id);
 
 CREATE TABLE IF NOT EXISTS remote_media_cache (
     media_origin VARCHAR(150), -- The remote HS the media came from.
     media_id VARCHAR(150), -- The id used to refer to the media on that server.
     media_type VARCHAR(150), -- The MIME-type of the media.
-    created_ts BIGINT UNSIGNED, -- When the content was uploaded in ms.
+    created_ts BIGINT, -- When the content was uploaded in ms.
     upload_name VARCHAR(150), -- The name the media was uploaded with.
     media_length INTEGER, -- Length of the media in bytes.
     filesystem_id VARCHAR(150), -- The name used to store the media on disk.
@@ -64,5 +64,5 @@ CREATE TABLE IF NOT EXISTS remote_media_cache_thumbnails (
      )
 );
 
-CREATE INDEX IF NOT EXISTS remote_media_cache_thumbnails_media_id
+CREATE INDEX remote_media_cache_thumbnails_media_id
     ON remote_media_cache_thumbnails (media_id);
diff --git a/synapse/storage/schema/full_schemas/16/presence.sql b/synapse/storage/schema/full_schemas/16/presence.sql
index fbe5b0af6c..9c41be296e 100644
--- a/synapse/storage/schema/full_schemas/16/presence.sql
+++ b/synapse/storage/schema/full_schemas/16/presence.sql
@@ -16,7 +16,7 @@ CREATE TABLE IF NOT EXISTS presence(
   user_id VARCHAR(150) NOT NULL,
   state VARCHAR(20),
   status_msg VARCHAR(150),
-  mtime BIGINT UNSIGNED, -- miliseconds since last state change
+  mtime BIGINT, -- miliseconds since last state change
   UNIQUE (user_id)
 );
 
@@ -37,4 +37,4 @@ CREATE TABLE IF NOT EXISTS presence_list(
   UNIQUE (user_id, observed_user_id)
 );
 
-CREATE INDEX IF NOT EXISTS presence_list_user_id ON presence_list (user_id);
+CREATE INDEX presence_list_user_id ON presence_list (user_id);
diff --git a/synapse/storage/schema/full_schemas/16/push.sql b/synapse/storage/schema/full_schemas/16/push.sql
index 33300736f9..5c0c7bc201 100644
--- a/synapse/storage/schema/full_schemas/16/push.sql
+++ b/synapse/storage/schema/full_schemas/16/push.sql
@@ -22,52 +22,52 @@ CREATE TABLE IF NOT EXISTS rejections(
 
 -- Push notification endpoints that users have configured
 CREATE TABLE IF NOT EXISTS pushers (
-  id BIGINT UNSIGNED PRIMARY KEY,
+  id BIGINT PRIMARY KEY,
   user_name VARCHAR(150) NOT NULL,
   profile_tag VARCHAR(32) NOT NULL,
   kind VARCHAR(8) NOT NULL,
   app_id VARCHAR(64) NOT NULL,
   app_display_name VARCHAR(64) NOT NULL,
   device_display_name VARCHAR(128) NOT NULL,
-  pushkey VARBINARY(512) NOT NULL,
-  ts BIGINT UNSIGNED NOT NULL,
+  pushkey bytea NOT NULL,
+  ts BIGINT NOT NULL,
   lang VARCHAR(8),
-  data LONGBLOB,
+  data bytea,
   last_token TEXT,
-  last_success BIGINT UNSIGNED,
-  failing_since BIGINT UNSIGNED,
+  last_success BIGINT,
+  failing_since BIGINT,
   UNIQUE (app_id, pushkey)
 );
 
 CREATE TABLE IF NOT EXISTS push_rules (
-  id BIGINT UNSIGNED PRIMARY KEY,
+  id BIGINT PRIMARY KEY,
   user_name VARCHAR(150) NOT NULL,
   rule_id VARCHAR(150) NOT NULL,
-  priority_class TINYINT NOT NULL,
+  priority_class SMALLINT NOT NULL,
   priority INTEGER NOT NULL DEFAULT 0,
   conditions VARCHAR(150) NOT NULL,
   actions VARCHAR(150) NOT NULL,
   UNIQUE(user_name, rule_id)
 );
 
-CREATE INDEX IF NOT EXISTS push_rules_user_name on push_rules (user_name);
+CREATE INDEX push_rules_user_name on push_rules (user_name);
 
 CREATE TABLE IF NOT EXISTS user_filters(
   user_id VARCHAR(150),
-  filter_id BIGINT UNSIGNED,
-  filter_json LONGBLOB
+  filter_id BIGINT,
+  filter_json bytea
 );
 
-CREATE INDEX IF NOT EXISTS user_filters_by_user_id_filter_id ON user_filters(
+CREATE INDEX user_filters_by_user_id_filter_id ON user_filters(
     user_id, filter_id
 );
 
 CREATE TABLE IF NOT EXISTS push_rules_enable (
-  id BIGINT UNSIGNED PRIMARY KEY,
+  id BIGINT PRIMARY KEY,
   user_name VARCHAR(150) NOT NULL,
   rule_id VARCHAR(150) NOT NULL,
-  enabled TINYINT,
+  enabled SMALLINT,
   UNIQUE(user_name, rule_id)
 );
 
-CREATE INDEX IF NOT EXISTS push_rules_enable_user_name on push_rules_enable (user_name);
+CREATE INDEX push_rules_enable_user_name on push_rules_enable (user_name);
diff --git a/synapse/storage/schema/full_schemas/16/redactions.sql b/synapse/storage/schema/full_schemas/16/redactions.sql
index b81451eab4..492fd22033 100644
--- a/synapse/storage/schema/full_schemas/16/redactions.sql
+++ b/synapse/storage/schema/full_schemas/16/redactions.sql
@@ -18,5 +18,5 @@ CREATE TABLE IF NOT EXISTS redactions (
     UNIQUE (event_id)
 );
 
-CREATE INDEX IF NOT EXISTS redactions_event_id ON redactions (event_id);
-CREATE INDEX IF NOT EXISTS redactions_redacts ON redactions (redacts);
+CREATE INDEX redactions_event_id ON redactions (event_id);
+CREATE INDEX redactions_redacts ON redactions (redacts);
diff --git a/synapse/storage/schema/full_schemas/16/state.sql b/synapse/storage/schema/full_schemas/16/state.sql
index 8c51610396..3c54595e64 100644
--- a/synapse/storage/schema/full_schemas/16/state.sql
+++ b/synapse/storage/schema/full_schemas/16/state.sql
@@ -14,7 +14,7 @@
  */
 
 CREATE TABLE IF NOT EXISTS state_groups(
-    id BIGINT UNSIGNED PRIMARY KEY,
+    id BIGINT PRIMARY KEY,
     room_id VARCHAR(150) NOT NULL,
     event_id VARCHAR(150) NOT NULL
 );
@@ -33,8 +33,8 @@ CREATE TABLE IF NOT EXISTS event_to_state_groups(
     UNIQUE (event_id)
 );
 
-CREATE INDEX IF NOT EXISTS state_groups_id ON state_groups(id);
+CREATE INDEX state_groups_id ON state_groups(id);
 
-CREATE INDEX IF NOT EXISTS state_groups_state_id ON state_groups_state(state_group);
-CREATE INDEX IF NOT EXISTS state_groups_state_tuple ON state_groups_state(room_id, type, state_key);
-CREATE INDEX IF NOT EXISTS event_to_state_groups_id ON event_to_state_groups(event_id);
+CREATE INDEX state_groups_state_id ON state_groups_state(state_group);
+CREATE INDEX state_groups_state_tuple ON state_groups_state(room_id, type, state_key);
+CREATE INDEX event_to_state_groups_id ON event_to_state_groups(event_id);
diff --git a/synapse/storage/schema/full_schemas/16/transactions.sql b/synapse/storage/schema/full_schemas/16/transactions.sql
index f381e67603..bc64064936 100644
--- a/synapse/storage/schema/full_schemas/16/transactions.sql
+++ b/synapse/storage/schema/full_schemas/16/transactions.sql
@@ -16,32 +16,32 @@
 CREATE TABLE IF NOT EXISTS received_transactions(
     transaction_id VARCHAR(150),
     origin VARCHAR(150),
-    ts BIGINT UNSIGNED,
+    ts BIGINT,
     response_code INTEGER,
-    response_json LONGBLOB,
-    has_been_referenced BOOL default 0, -- Whether thishas been referenced by a prev_tx
+    response_json bytea,
+    has_been_referenced smallint default 0, -- Whether thishas been referenced by a prev_tx
     UNIQUE (transaction_id, origin)
 );
 
-CREATE INDEX IF NOT EXISTS transactions_have_ref ON received_transactions(origin, has_been_referenced);-- WHERE has_been_referenced = 0;
+CREATE INDEX transactions_have_ref ON received_transactions(origin, has_been_referenced);-- WHERE has_been_referenced = 0;
 
 
 -- Stores what transactions we've sent, what their response was (if we got one) and whether we have
 -- since referenced the transaction in another outgoing transaction
 CREATE TABLE IF NOT EXISTS sent_transactions(
-    id BIGINT UNSIGNED PRIMARY KEY, -- This is used to apply insertion ordering
+    id BIGINT PRIMARY KEY, -- This is used to apply insertion ordering
     transaction_id VARCHAR(150),
     destination VARCHAR(150),
     response_code INTEGER DEFAULT 0,
-    response_json LONGBLOB,
-    ts BIGINT UNSIGNED
+    response_json bytea,
+    ts BIGINT
 );
 
-CREATE INDEX IF NOT EXISTS sent_transaction_dest ON sent_transactions(destination);
-CREATE INDEX IF NOT EXISTS sent_transaction_txn_id ON sent_transactions(transaction_id);
+CREATE INDEX sent_transaction_dest ON sent_transactions(destination);
+CREATE INDEX sent_transaction_txn_id ON sent_transactions(transaction_id);
 -- So that we can do an efficient look up of all transactions that have yet to be successfully
 -- sent.
-CREATE INDEX IF NOT EXISTS sent_transaction_sent ON sent_transactions(response_code);
+CREATE INDEX sent_transaction_sent ON sent_transactions(response_code);
 
 
 -- For sent transactions only.
@@ -53,11 +53,11 @@ CREATE TABLE IF NOT EXISTS transaction_id_to_pdu(
     UNIQUE (transaction_id, destination)
 );
 
-CREATE INDEX IF NOT EXISTS transaction_id_to_pdu_dest ON transaction_id_to_pdu(destination);
+CREATE INDEX transaction_id_to_pdu_dest ON transaction_id_to_pdu(destination);
 
 -- To track destination health
 CREATE TABLE IF NOT EXISTS destinations(
     destination VARCHAR(150) PRIMARY KEY,
-    retry_last_ts BIGINT UNSIGNED,
+    retry_last_ts BIGINT,
     retry_interval INTEGER
 );
diff --git a/synapse/storage/schema/full_schemas/16/users.sql b/synapse/storage/schema/full_schemas/16/users.sql
index d0011c04b4..267284d07d 100644
--- a/synapse/storage/schema/full_schemas/16/users.sql
+++ b/synapse/storage/schema/full_schemas/16/users.sql
@@ -15,17 +15,17 @@
 CREATE TABLE IF NOT EXISTS users(
     name VARCHAR(150),
     password_hash VARCHAR(150),
-    creation_ts BIGINT UNSIGNED,
-    admin BOOL DEFAULT 0 NOT NULL,
+    creation_ts BIGINT,
+    admin SMALLINT DEFAULT 0 NOT NULL,
     UNIQUE(name)
 );
 
 CREATE TABLE IF NOT EXISTS access_tokens(
-    id BIGINT UNSIGNED PRIMARY KEY,
+    id BIGINT PRIMARY KEY,
     user_id VARCHAR(150) NOT NULL,
     device_id VARCHAR(150),
     token VARCHAR(150) NOT NULL,
-    last_used BIGINT UNSIGNED,
+    last_used BIGINT,
     UNIQUE(token)
 );
 
@@ -35,7 +35,7 @@ CREATE TABLE IF NOT EXISTS user_ips (
     device_id VARCHAR(150),
     ip VARCHAR(150) NOT NULL,
     user_agent VARCHAR(150) NOT NULL,
-    last_seen BIGINT UNSIGNED NOT NULL
+    last_seen BIGINT NOT NULL
 );
 
 CREATE INDEX IF NOT EXISTS user_ips_user ON user_ips(user);
diff --git a/synapse/storage/schema/schema_version.sql b/synapse/storage/schema/schema_version.sql
index e7fa6fe569..d9494611e0 100644
--- a/synapse/storage/schema/schema_version.sql
+++ b/synapse/storage/schema/schema_version.sql
@@ -14,16 +14,14 @@
  */
 
 CREATE TABLE IF NOT EXISTS schema_version(
-    `Lock` CHAR(1) NOT NULL DEFAULT 'X' UNIQUE,  -- Makes sure this table only has one row.
-    `version` INTEGER NOT NULL,
-    `upgraded` BOOL NOT NULL,  -- Whether we reached this version from an upgrade or an initial schema.
-    CHECK (`Lock`='X')
+    Lock CHAR(1) NOT NULL DEFAULT 'X' UNIQUE,  -- Makes sure this table only has one row.
+    version INTEGER NOT NULL,
+    upgraded BOOL NOT NULL,  -- Whether we reached this version from an upgrade or an initial schema.
+    CHECK (Lock='X')
 );
 
 CREATE TABLE IF NOT EXISTS applied_schema_deltas(
-    `version` INTEGER NOT NULL,
-    `file` VARCHAR(150) NOT NULL,
+    version INTEGER NOT NULL,
+    file VARCHAR(150) NOT NULL,
     UNIQUE(version, file)
 );
-
-CREATE INDEX IF NOT EXISTS schema_deltas_ver ON applied_schema_deltas(version);
diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py
index 57c2e4dfeb..df6de7cbcd 100644
--- a/synapse/storage/stream.py
+++ b/synapse/storage/stream.py
@@ -240,7 +240,7 @@ class StreamStore(SQLBaseStore):
 
         sql = (
             "SELECT e.event_id, e.stream_ordering FROM events AS e WHERE "
-            "(e.outlier = 0 AND (room_id IN (%(current)s)) OR "
+            "(e.outlier = ? AND (room_id IN (%(current)s)) OR "
             "(event_id IN (%(invites)s))) "
             "AND e.stream_ordering > ? AND e.stream_ordering <= ? "
             "ORDER BY stream_ordering ASC LIMIT %(limit)d "
@@ -251,7 +251,7 @@ class StreamStore(SQLBaseStore):
         }
 
         def f(txn):
-            txn.execute(sql, (user_id, user_id, from_id.stream, to_id.stream,))
+            txn.execute(sql, (False, user_id, user_id, from_id.stream, to_id.stream,))
 
             rows = self.cursor_to_dict(txn)
 
@@ -283,7 +283,7 @@ class StreamStore(SQLBaseStore):
         # Tokens really represent positions between elements, but we use
         # the convention of pointing to the event before the gap. Hence
         # we have a bit of asymmetry when it comes to equalities.
-        args = [room_id]
+        args = [False, room_id]
         if direction == 'b':
             order = "DESC"
             bounds = _StreamToken.parse(from_key).upper_bound()
@@ -307,7 +307,7 @@ class StreamStore(SQLBaseStore):
 
         sql = (
             "SELECT * FROM events"
-            " WHERE outlier = 0 AND room_id = ? AND %(bounds)s"
+            " WHERE outlier = ? AND room_id = ? AND %(bounds)s"
             " ORDER BY topological_ordering %(order)s,"
             " stream_ordering %(order)s %(limit)s"
         ) % {
@@ -358,7 +358,7 @@ class StreamStore(SQLBaseStore):
             sql = (
                 "SELECT stream_ordering, topological_ordering, event_id"
                 " FROM events"
-                " WHERE room_id = ? AND stream_ordering <= ? AND outlier = 0"
+                " WHERE room_id = ? AND stream_ordering <= ? AND outlier = ?"
                 " ORDER BY topological_ordering DESC, stream_ordering DESC"
                 " LIMIT ?"
             )
@@ -368,17 +368,17 @@ class StreamStore(SQLBaseStore):
                 "SELECT stream_ordering, topological_ordering, event_id"
                 " FROM events"
                 " WHERE room_id = ? AND stream_ordering > ?"
-                " AND stream_ordering <= ? AND outlier = 0"
+                " AND stream_ordering <= ? AND outlier = ?"
                 " ORDER BY topological_ordering DESC, stream_ordering DESC"
                 " LIMIT ?"
             )
 
         def get_recent_events_for_room_txn(txn):
             if from_token is None:
-                txn.execute(sql, (room_id, end_token.stream, limit,))
+                txn.execute(sql, (room_id, end_token.stream, False, limit,))
             else:
                 txn.execute(sql, (
-                    room_id, from_token.stream, end_token.stream, limit
+                    room_id, from_token.stream, end_token.stream, False, limit
                 ))
 
             rows = self.cursor_to_dict(txn)
-- 
cgit 1.4.1


From ded41289655d083194061678c24d3e5551e64a7d Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Wed, 15 Apr 2015 10:24:24 +0100
Subject: Use True for True rather than 1

---
 synapse/storage/events.py | 12 +++++++-----
 1 file changed, 7 insertions(+), 5 deletions(-)

(limited to 'synapse/storage/events.py')

diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 9fe2effb4b..0373d152b2 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -269,13 +269,15 @@ class EventsStore(SQLBaseStore):
             "INSERT INTO events"
             " (stream_ordering, topological_ordering, event_id, type,"
             " room_id, content, processed, outlier, depth)"
-            " VALUES (%s,?,?,?,?,?,?,?,?)"
-        ) % (stream_ordering,)
+            " VALUES (?,?,?,?,?,?,?,?,?)"
+        )
 
         txn.execute(
             sql,
-            (event.depth, event.event_id, event.type, event.room_id,
-             content, True, outlier, event.depth)
+            (
+                stream_ordering, event.depth, event.event_id, event.type,
+                event.room_id, content, True, outlier, event.depth
+            )
         )
 
         if context.rejected:
@@ -321,7 +323,7 @@ class EventsStore(SQLBaseStore):
                         "event_id": event.event_id,
                         "prev_event_id": e_id,
                         "room_id": event.room_id,
-                        "is_state": 1,
+                        "is_state": True,
                     },
                 )
 
-- 
cgit 1.4.1


From b8092fbc82edb3c7d8aa09f0756fa853ad6a6ad8 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 16 Apr 2015 11:17:52 +0100
Subject: Go back to storing JSON in TEXT

---
 synapse/storage/_base.py                                 |  3 ---
 synapse/storage/appservice.py                            |  4 +---
 synapse/storage/engines/postgres.py                      |  3 ---
 synapse/storage/engines/sqlite3.py                       |  5 -----
 synapse/storage/events.py                                | 16 ++++++++--------
 synapse/storage/profile.py                               |  5 +----
 synapse/storage/registration.py                          |  5 -----
 synapse/storage/schema/delta/15/appservice_txns.sql      |  2 +-
 synapse/storage/schema/full_schemas/11/im.sql            |  8 ++++----
 .../schema/full_schemas/16/application_services.sql      |  2 +-
 synapse/storage/schema/full_schemas/16/im.sql            |  8 ++++----
 11 files changed, 20 insertions(+), 41 deletions(-)

(limited to 'synapse/storage/events.py')

diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 5ec1d2613e..f5952d1fc0 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -818,9 +818,6 @@ class SQLBaseStore(object):
 
         internal_metadata, js, redacted, rejected_reason = res
 
-        internal_metadata = self.database_engine.load_unicode(internal_metadata)
-        js = self.database_engine.load_unicode(js)
-
         start_time = update_counter("select_event", start_time)
 
         result = self._get_event_from_row_txn(
diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py
index 40e05b3635..63d1af4e86 100644
--- a/synapse/storage/appservice.py
+++ b/synapse/storage/appservice.py
@@ -366,9 +366,7 @@ class ApplicationServiceTransactionStore(SQLBaseStore):
         new_txn_id = max(highest_txn_id, last_txn_id) + 1
 
         # Insert new txn into txn table
-        event_ids = buffer(
-            json.dumps([e.event_id for e in events]).encode("utf8")
-        )
+        event_ids = json.dumps([e.event_id for e in events])
         txn.execute(
             "INSERT INTO application_services_txns(as_id, txn_id, event_ids) "
             "VALUES(?,?,?)",
diff --git a/synapse/storage/engines/postgres.py b/synapse/storage/engines/postgres.py
index 457c1f70a5..6f75245fa7 100644
--- a/synapse/storage/engines/postgres.py
+++ b/synapse/storage/engines/postgres.py
@@ -39,6 +39,3 @@ class PostgresEngine(object):
         if isinstance(error, self.module.DatabaseError):
             return error.pgcode in ["40001", "40P01"]
         return False
-
-    def load_unicode(self, v):
-        return bytes(v).decode("UTF8")
diff --git a/synapse/storage/engines/sqlite3.py b/synapse/storage/engines/sqlite3.py
index 389df35eb5..dd0d8e0e0f 100644
--- a/synapse/storage/engines/sqlite3.py
+++ b/synapse/storage/engines/sqlite3.py
@@ -37,8 +37,3 @@ class Sqlite3Engine(object):
 
     def is_deadlock(self, error):
         return False
-
-    def load_unicode(self, v):
-        if isinstance(v, types.UnicodeType):
-            return v
-        return bytes(v).decode("UTF8")
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 0373d152b2..7dbf7a396a 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -168,7 +168,7 @@ class EventsStore(SQLBaseStore):
 
         metadata_json = encode_canonical_json(
             event.internal_metadata.get_dict()
-        )
+        ).decode("UTF-8")
 
         # If we have already persisted this event, we don't need to do any
         # more processing.
@@ -184,7 +184,7 @@ class EventsStore(SQLBaseStore):
                 )
                 txn.execute(
                     sql,
-                    (buffer(metadata_json), event.event_id,)
+                    (metadata_json, event.event_id,)
                 )
 
                 sql = (
@@ -229,14 +229,14 @@ class EventsStore(SQLBaseStore):
             values={
                 "event_id": event.event_id,
                 "room_id": event.room_id,
-                "internal_metadata": buffer(metadata_json),
-                "json": buffer(encode_canonical_json(event_dict)),
+                "internal_metadata": metadata_json,
+                "json": encode_canonical_json(event_dict).decode("UTF-8"),
             },
         )
 
-        content = buffer(encode_canonical_json(
+        content = encode_canonical_json(
             event.content
-        ))
+        ).decode("UTF-8")
 
         vals = {
             "topological_ordering": event.depth,
@@ -261,9 +261,9 @@ class EventsStore(SQLBaseStore):
             ]
         }
 
-        vals["unrecognized_keys"] = buffer(encode_canonical_json(
+        vals["unrecognized_keys"] = encode_canonical_json(
             unrec
-        ))
+        ).decode("UTF-8")
 
         sql = (
             "INSERT INTO events"
diff --git a/synapse/storage/profile.py b/synapse/storage/profile.py
index e33963d0b4..047698aa13 100644
--- a/synapse/storage/profile.py
+++ b/synapse/storage/profile.py
@@ -35,16 +35,13 @@ class ProfileStore(SQLBaseStore):
             desc="get_profile_displayname",
         )
 
-        if name:
-            name = self.database_engine.load_unicode(name)
-
         defer.returnValue(name)
 
     def set_profile_displayname(self, user_localpart, new_displayname):
         return self._simple_update_one(
             table="profiles",
             keyvalues={"user_id": user_localpart},
-            updatevalues={"displayname": new_displayname.encode("utf8")},
+            updatevalues={"displayname": new_displayname},
             desc="set_profile_displayname",
         )
 
diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py
index 8a63fe4691..2a5c5080e4 100644
--- a/synapse/storage/registration.py
+++ b/synapse/storage/registration.py
@@ -98,11 +98,6 @@ class RegistrationStore(SQLBaseStore):
             allow_none=True,
         )
 
-        if user_info:
-            user_info["password_hash"] = self.database_engine.load_unicode(
-                user_info["password_hash"]
-            )
-
         defer.returnValue(user_info)
 
     @cached()
diff --git a/synapse/storage/schema/delta/15/appservice_txns.sql b/synapse/storage/schema/delta/15/appservice_txns.sql
index ddea8fc693..1c3324f415 100644
--- a/synapse/storage/schema/delta/15/appservice_txns.sql
+++ b/synapse/storage/schema/delta/15/appservice_txns.sql
@@ -22,7 +22,7 @@ CREATE TABLE IF NOT EXISTS application_services_state(
 CREATE TABLE IF NOT EXISTS application_services_txns(
     as_id VARCHAR(150) NOT NULL,
     txn_id INTEGER NOT NULL,
-    event_ids LONGBLOB NOT NULL,
+    event_ids TEXT NOT NULL,
     UNIQUE(as_id, txn_id)
 );
 
diff --git a/synapse/storage/schema/full_schemas/11/im.sql b/synapse/storage/schema/full_schemas/11/im.sql
index e9e09214d7..addbec5885 100644
--- a/synapse/storage/schema/full_schemas/11/im.sql
+++ b/synapse/storage/schema/full_schemas/11/im.sql
@@ -19,8 +19,8 @@ CREATE TABLE IF NOT EXISTS events(
     event_id VARCHAR(150) NOT NULL,
     type VARCHAR(150) NOT NULL,
     room_id VARCHAR(150) NOT NULL,
-    content bytea NOT NULL,
-    unrecognized_keys bytea,
+    content TEXT NOT NULL,
+    unrecognized_keys TEXT,
     processed BOOL NOT NULL,
     outlier BOOL NOT NULL,
     depth BIGINT DEFAULT 0 NOT NULL,
@@ -35,8 +35,8 @@ CREATE INDEX events_room_id ON events (room_id);
 CREATE TABLE IF NOT EXISTS event_json(
     event_id VARCHAR(150) NOT NULL,
     room_id VARCHAR(150) NOT NULL,
-    internal_metadata bytea NOT NULL,
-    json bytea NOT NULL,
+    internal_metadata TEXT NOT NULL,
+    json TEXT NOT NULL,
     UNIQUE (event_id)
 );
 
diff --git a/synapse/storage/schema/full_schemas/16/application_services.sql b/synapse/storage/schema/full_schemas/16/application_services.sql
index f08c5bcf76..5d63d57d59 100644
--- a/synapse/storage/schema/full_schemas/16/application_services.sql
+++ b/synapse/storage/schema/full_schemas/16/application_services.sql
@@ -39,7 +39,7 @@ CREATE TABLE IF NOT EXISTS application_services_state(
 CREATE TABLE IF NOT EXISTS application_services_txns(
     as_id VARCHAR(150) NOT NULL,
     txn_id INTEGER NOT NULL,
-    event_ids bytea NOT NULL,
+    event_ids TEXT NOT NULL,
     UNIQUE(as_id, txn_id)
 );
 
diff --git a/synapse/storage/schema/full_schemas/16/im.sql b/synapse/storage/schema/full_schemas/16/im.sql
index 17e4c949b9..5b4b494484 100644
--- a/synapse/storage/schema/full_schemas/16/im.sql
+++ b/synapse/storage/schema/full_schemas/16/im.sql
@@ -19,8 +19,8 @@ CREATE TABLE IF NOT EXISTS events(
     event_id VARCHAR(150) NOT NULL,
     type VARCHAR(150) NOT NULL,
     room_id VARCHAR(150) NOT NULL,
-    content bytea NOT NULL,
-    unrecognized_keys bytea,
+    content TEXT NOT NULL,
+    unrecognized_keys TEXT,
     processed BOOL NOT NULL,
     outlier BOOL NOT NULL,
     depth BIGINT DEFAULT 0 NOT NULL,
@@ -39,8 +39,8 @@ CREATE INDEX events_order_room ON events (
 CREATE TABLE IF NOT EXISTS event_json(
     event_id VARCHAR(150) NOT NULL,
     room_id VARCHAR(150) NOT NULL,
-    internal_metadata bytea NOT NULL,
-    json bytea NOT NULL,
+    internal_metadata TEXT NOT NULL,
+    json TEXT NOT NULL,
     UNIQUE (event_id)
 );
 
-- 
cgit 1.4.1


From 2732be83d9e883184f4a783fb7ba15487f30c20d Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 27 Apr 2015 13:22:30 +0100
Subject: Shuffle operations so that locking upsert happens last in the txn.
 This ensures the lock is held for the least amount of time possible.

---
 synapse/storage/_base.py            |  3 ++
 synapse/storage/engines/postgres.py |  3 ++
 synapse/storage/engines/sqlite3.py  |  3 ++
 synapse/storage/events.py           | 82 ++++++++++++++++++-------------------
 synapse/storage/transactions.py     | 21 +++-------
 5 files changed, 56 insertions(+), 56 deletions(-)

(limited to 'synapse/storage/events.py')

diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 874d41447a..6017c2a6e8 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -466,6 +466,9 @@ class SQLBaseStore(object):
         )
 
     def _simple_upsert_txn(self, txn, table, keyvalues, values, insertion_values={}):
+        # We need to lock the table :(
+        self.database_engine.lock_table(txn, table)
+
         # Try to update
         sql = "UPDATE %s SET %s WHERE %s" % (
             table,
diff --git a/synapse/storage/engines/postgres.py b/synapse/storage/engines/postgres.py
index 00dbae7b60..b8cca9b187 100644
--- a/synapse/storage/engines/postgres.py
+++ b/synapse/storage/engines/postgres.py
@@ -42,3 +42,6 @@ class PostgresEngine(object):
 
     def is_connection_closed(self, conn):
         return bool(conn)
+
+    def lock_table(self, txn, table):
+        txn.execute("LOCK TABLE %s in EXCLUSIVE MODE" % (table,))
diff --git a/synapse/storage/engines/sqlite3.py b/synapse/storage/engines/sqlite3.py
index 39828a597c..f62d5d1205 100644
--- a/synapse/storage/engines/sqlite3.py
+++ b/synapse/storage/engines/sqlite3.py
@@ -38,3 +38,6 @@ class Sqlite3Engine(object):
 
     def is_connection_closed(self, conn):
         return False
+
+    def lock_table(self, txn, table):
+        return
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 7dbf7a396a..a3c260ddc4 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -283,6 +283,35 @@ class EventsStore(SQLBaseStore):
         if context.rejected:
             self._store_rejections_txn(txn, event.event_id, context.rejected)
 
+        for hash_alg, hash_base64 in event.hashes.items():
+            hash_bytes = decode_base64(hash_base64)
+            self._store_event_content_hash_txn(
+                txn, event.event_id, hash_alg, hash_bytes,
+            )
+
+        for prev_event_id, prev_hashes in event.prev_events:
+            for alg, hash_base64 in prev_hashes.items():
+                hash_bytes = decode_base64(hash_base64)
+                self._store_prev_event_hash_txn(
+                    txn, event.event_id, prev_event_id, alg, hash_bytes
+                )
+
+        for auth_id, _ in event.auth_events:
+            self._simple_insert_txn(
+                txn,
+                table="event_auth",
+                values={
+                    "event_id": event.event_id,
+                    "room_id": event.room_id,
+                    "auth_id": auth_id,
+                },
+            )
+
+        (ref_alg, ref_hash_bytes) = compute_event_reference_hash(event)
+        self._store_event_reference_hash_txn(
+            txn, event.event_id, ref_alg, ref_hash_bytes
+        )
+
         if event.is_state():
             vals = {
                 "event_id": event.event_id,
@@ -301,20 +330,6 @@ class EventsStore(SQLBaseStore):
                 vals,
             )
 
-            if is_new_state and not context.rejected:
-                self._simple_upsert_txn(
-                    txn,
-                    "current_state_events",
-                    keyvalues={
-                        "room_id": event.room_id,
-                        "type": event.type,
-                        "state_key": event.state_key,
-                    },
-                    values={
-                        "event_id": event.event_id,
-                    }
-                )
-
             for e_id, h in event.prev_state:
                 self._simple_insert_txn(
                     txn,
@@ -327,35 +342,20 @@ class EventsStore(SQLBaseStore):
                     },
                 )
 
-        for hash_alg, hash_base64 in event.hashes.items():
-            hash_bytes = decode_base64(hash_base64)
-            self._store_event_content_hash_txn(
-                txn, event.event_id, hash_alg, hash_bytes,
-            )
-
-        for prev_event_id, prev_hashes in event.prev_events:
-            for alg, hash_base64 in prev_hashes.items():
-                hash_bytes = decode_base64(hash_base64)
-                self._store_prev_event_hash_txn(
-                    txn, event.event_id, prev_event_id, alg, hash_bytes
+            if is_new_state and not context.rejected:
+                self._simple_upsert_txn(
+                    txn,
+                    "current_state_events",
+                    keyvalues={
+                        "room_id": event.room_id,
+                        "type": event.type,
+                        "state_key": event.state_key,
+                    },
+                    values={
+                        "event_id": event.event_id,
+                    }
                 )
 
-        for auth_id, _ in event.auth_events:
-            self._simple_insert_txn(
-                txn,
-                table="event_auth",
-                values={
-                    "event_id": event.event_id,
-                    "room_id": event.room_id,
-                    "auth_id": auth_id,
-                },
-            )
-
-        (ref_alg, ref_hash_bytes) = compute_event_reference_hash(event)
-        self._store_event_reference_hash_txn(
-            txn, event.event_id, ref_alg, ref_hash_bytes
-        )
-
     def _store_redaction(self, txn, event):
         # invalidate the cache for the redacted event
         self._invalidate_get_event_cache(event.redacts)
diff --git a/synapse/storage/transactions.py b/synapse/storage/transactions.py
index 7e3add5280..89dd7d8947 100644
--- a/synapse/storage/transactions.py
+++ b/synapse/storage/transactions.py
@@ -76,25 +76,16 @@ class TransactionStore(SQLBaseStore):
             response_json (str)
         """
 
-        return self.runInteraction(
-            "set_received_txn_response",
-            self._set_received_txn_response,
-            transaction_id, origin, code, response_dict
-        )
-
-    def _set_received_txn_response(self, txn, transaction_id, origin, code,
-                                   response_json):
-        self._simple_upsert_txn(
-            txn,
+        return self._simple_insert(
             table=ReceivedTransactionsTable.table_name,
-            keyvalues={
+            values={
                 "transaction_id": transaction_id,
                 "origin": origin,
-            },
-            values={
                 "response_code": code,
-                "response_json": response_json,
-            }
+                "response_json": response_dict,
+            },
+            or_ignore=True,
+            desc="set_received_txn_response",
         )
 
     def prep_send_transaction(self, transaction_id, destination,
-- 
cgit 1.4.1