From d7a0496f3ec534076121632352f44733253e1e16 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 19 Mar 2015 15:59:48 +0000 Subject: Convert storage layer to be mysql compatible --- synapse/storage/roommember.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'synapse/storage/roommember.py') diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py index 65ffb4627f..e8ede14cd7 100644 --- a/synapse/storage/roommember.py +++ b/synapse/storage/roommember.py @@ -68,7 +68,7 @@ class RoomMemberStore(SQLBaseStore): # Update room hosts table if event.membership == Membership.JOIN: sql = ( - "INSERT OR IGNORE INTO room_hosts (room_id, host) " + "REPLACE INTO room_hosts (room_id, host) " "VALUES (?, ?)" ) txn.execute(sql, (event.room_id, domain)) -- cgit 1.5.1 From dc0c989ef43ecfe3b4159e0cf16c50a7f38c1f20 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 20 Mar 2015 15:59:18 +0000 Subject: Give sensible names for '_simple_...' transactions --- synapse/storage/__init__.py | 4 +++- synapse/storage/_base.py | 37 +++++++++++++++++++++---------------- synapse/storage/directory.py | 7 ++++++- synapse/storage/filtering.py | 1 + synapse/storage/media_repository.py | 20 ++++++++++++++------ synapse/storage/presence.py | 11 +++++++++++ synapse/storage/profile.py | 5 +++++ synapse/storage/push_rule.py | 9 ++++++--- synapse/storage/pusher.py | 16 +++++++++++----- synapse/storage/registration.py | 9 +++++++-- synapse/storage/rejections.py | 3 ++- synapse/storage/room.py | 6 ++++-- synapse/storage/roommember.py | 3 ++- 13 files changed, 93 insertions(+), 38 deletions(-) (limited to 'synapse/storage/roommember.py') diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 4295f7348e..76e7bdfaed 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -83,7 +83,8 @@ class DataStore(RoomMemberStore, RoomStore, "ip": ip, "user_agent": user_agent, "last_seen": int(self._clock.time_msec()), - } + }, + desc="insert_client_ip", ) def get_user_ip_and_agents(self, user): @@ -93,6 +94,7 @@ class DataStore(RoomMemberStore, RoomStore, retcols=[ "device_id", "access_token", "ip", "user_agent", "last_seen" ], + desc="get_user_ip_and_agents", ) diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index a6f94de01e..0aab9a8af4 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -321,7 +321,8 @@ class SQLBaseStore(object): # "Simple" SQL API methods that operate on a single table with no JOINs, # no complex WHERE clauses, just a dict of values for columns. - def _simple_insert(self, table, values, or_replace=False, or_ignore=False): + def _simple_insert(self, table, values, or_replace=False, or_ignore=False, + desc="_simple_insert"): """Executes an INSERT query on the named table. Args: @@ -330,7 +331,7 @@ class SQLBaseStore(object): or_replace : bool; if True performs an INSERT OR REPLACE """ return self.runInteraction( - "_simple_insert", + desc, self._simple_insert_txn, table, values, or_replace=or_replace, or_ignore=or_ignore, ) @@ -354,7 +355,7 @@ class SQLBaseStore(object): txn.execute(sql, values.values()) return txn.lastrowid - def _simple_upsert(self, table, keyvalues, values): + def _simple_upsert(self, table, keyvalues, values, desc="_simple_upsert"): """ Args: table (str): The table to upsert into @@ -363,7 +364,7 @@ class SQLBaseStore(object): Returns: A deferred """ return self.runInteraction( - "_simple_upsert", + desc, self._simple_upsert_txn, table, keyvalues, values ) @@ -418,7 +419,8 @@ class SQLBaseStore(object): ) def _simple_select_one_onecol(self, table, keyvalues, retcol, - allow_none=False): + allow_none=False, + desc="_simple_select_one_onecol"): """Executes a SELECT query on the named table, which is expected to return a single row, returning a single column from it." @@ -428,7 +430,7 @@ class SQLBaseStore(object): retcol : string giving the name of the column to return """ return self.runInteraction( - "_simple_select_one_onecol", + desc, self._simple_select_one_onecol_txn, table, keyvalues, retcol, allow_none=allow_none, ) @@ -464,7 +466,8 @@ class SQLBaseStore(object): return [r[0] for r in txn.fetchall()] - def _simple_select_onecol(self, table, keyvalues, retcol): + def _simple_select_onecol(self, table, keyvalues, retcol, + desc="_simple_select_onecol"): """Executes a SELECT query on the named table, which returns a list comprising of the values of the named column from the selected rows. @@ -477,12 +480,13 @@ class SQLBaseStore(object): Deferred: Results in a list """ return self.runInteraction( - "_simple_select_onecol", + desc, self._simple_select_onecol_txn, table, keyvalues, retcol ) - def _simple_select_list(self, table, keyvalues, retcols): + def _simple_select_list(self, table, keyvalues, retcols, + desc="_simple_select_list"): """Executes a SELECT query on the named table, which may return zero or more rows, returning the result as a list of dicts. @@ -493,7 +497,7 @@ class SQLBaseStore(object): retcols : list of strings giving the names of the columns to return """ return self.runInteraction( - "_simple_select_list", + desc, self._simple_select_list_txn, table, keyvalues, retcols ) @@ -587,7 +591,8 @@ class SQLBaseStore(object): return dict(zip(retcols, row)) def _simple_selectupdate_one(self, table, keyvalues, updatevalues=None, - retcols=None, allow_none=False): + retcols=None, allow_none=False, + desc="_simple_selectupdate_one"): """ Combined SELECT then UPDATE.""" def func(txn): ret = None @@ -609,9 +614,9 @@ class SQLBaseStore(object): ) return ret - return self.runInteraction("_simple_selectupdate_one", func) + return self.runInteraction(desc, func) - def _simple_delete_one(self, table, keyvalues): + def _simple_delete_one(self, table, keyvalues, desc="_simple_delete_one"): """Executes a DELETE query on the named table, expecting to delete a single row. @@ -630,9 +635,9 @@ class SQLBaseStore(object): raise StoreError(404, "No row found") if txn.rowcount > 1: raise StoreError(500, "more than one row matched") - return self.runInteraction("_simple_delete_one", func) + return self.runInteraction(desc, func) - def _simple_delete(self, table, keyvalues): + def _simple_delete(self, table, keyvalues, desc="_simple_delete"): """Executes a DELETE query on the named table. Args: @@ -640,7 +645,7 @@ class SQLBaseStore(object): keyvalues : dict of column names and values to select the row with """ - return self.runInteraction("_simple_delete", self._simple_delete_txn) + return self.runInteraction(desc, self._simple_delete_txn) def _simple_delete_txn(self, txn, table, keyvalues): sql = "DELETE FROM %s WHERE %s" % ( diff --git a/synapse/storage/directory.py b/synapse/storage/directory.py index 68b7d59693..6672752fe0 100644 --- a/synapse/storage/directory.py +++ b/synapse/storage/directory.py @@ -48,6 +48,7 @@ class DirectoryStore(SQLBaseStore): {"room_alias": room_alias.to_string()}, "room_id", allow_none=True, + desc="get_association_from_room_alias", ) if not room_id: @@ -58,6 +59,7 @@ class DirectoryStore(SQLBaseStore): "room_alias_servers", {"room_alias": room_alias.to_string()}, "server", + desc="get_association_from_room_alias", ) if not servers: @@ -87,6 +89,7 @@ class DirectoryStore(SQLBaseStore): "room_alias": room_alias.to_string(), "room_id": room_id, }, + desc="create_room_alias_association", ) except sqlite3.IntegrityError: raise SynapseError( @@ -100,7 +103,8 @@ class DirectoryStore(SQLBaseStore): { "room_alias": room_alias.to_string(), "server": server, - } + }, + desc="create_room_alias_association", ) def delete_room_alias(self, room_alias): @@ -139,4 +143,5 @@ class DirectoryStore(SQLBaseStore): "room_aliases", {"room_id": room_id}, "room_alias", + desc="get_aliases_for_room", ) diff --git a/synapse/storage/filtering.py b/synapse/storage/filtering.py index 457a11fd02..8800116570 100644 --- a/synapse/storage/filtering.py +++ b/synapse/storage/filtering.py @@ -31,6 +31,7 @@ class FilteringStore(SQLBaseStore): }, retcol="filter_json", allow_none=False, + desc="get_user_filter", ) defer.returnValue(json.loads(def_json)) diff --git a/synapse/storage/media_repository.py b/synapse/storage/media_repository.py index 7101d2beec..7bf57234f6 100644 --- a/synapse/storage/media_repository.py +++ b/synapse/storage/media_repository.py @@ -32,6 +32,7 @@ class MediaRepositoryStore(SQLBaseStore): {"media_id": media_id}, ("media_type", "media_length", "upload_name", "created_ts"), allow_none=True, + desc="get_local_media", ) def store_local_media(self, media_id, media_type, time_now_ms, upload_name, @@ -45,7 +46,8 @@ class MediaRepositoryStore(SQLBaseStore): "upload_name": upload_name, "media_length": media_length, "user_id": user_id.to_string(), - } + }, + desc="store_local_media", ) def get_local_media_thumbnails(self, media_id): @@ -55,7 +57,8 @@ class MediaRepositoryStore(SQLBaseStore): ( "thumbnail_width", "thumbnail_height", "thumbnail_method", "thumbnail_type", "thumbnail_length", - ) + ), + desc="get_local_media_thumbnails", ) def store_local_thumbnail(self, media_id, thumbnail_width, @@ -70,7 +73,8 @@ class MediaRepositoryStore(SQLBaseStore): "thumbnail_method": thumbnail_method, "thumbnail_type": thumbnail_type, "thumbnail_length": thumbnail_length, - } + }, + desc="store_local_thumbnail", ) def get_cached_remote_media(self, origin, media_id): @@ -82,6 +86,7 @@ class MediaRepositoryStore(SQLBaseStore): "filesystem_id", ), allow_none=True, + desc="get_cached_remote_media", ) def store_cached_remote_media(self, origin, media_id, media_type, @@ -97,7 +102,8 @@ class MediaRepositoryStore(SQLBaseStore): "created_ts": time_now_ms, "upload_name": upload_name, "filesystem_id": filesystem_id, - } + }, + desc="store_cached_remote_media", ) def get_remote_media_thumbnails(self, origin, media_id): @@ -107,7 +113,8 @@ class MediaRepositoryStore(SQLBaseStore): ( "thumbnail_width", "thumbnail_height", "thumbnail_method", "thumbnail_type", "thumbnail_length", "filesystem_id", - ) + ), + desc="get_remote_media_thumbnails", ) def store_remote_media_thumbnail(self, origin, media_id, filesystem_id, @@ -125,5 +132,6 @@ class MediaRepositoryStore(SQLBaseStore): "thumbnail_type": thumbnail_type, "thumbnail_length": thumbnail_length, "filesystem_id": filesystem_id, - } + }, + desc="store_remote_media_thumbnail", ) diff --git a/synapse/storage/presence.py b/synapse/storage/presence.py index 0084d67e5b..87fba55439 100644 --- a/synapse/storage/presence.py +++ b/synapse/storage/presence.py @@ -21,6 +21,7 @@ class PresenceStore(SQLBaseStore): return self._simple_insert( table="presence", values={"user_id": user_localpart}, + desc="create_presence", ) def has_presence_state(self, user_localpart): @@ -29,6 +30,7 @@ class PresenceStore(SQLBaseStore): keyvalues={"user_id": user_localpart}, retcols=["user_id"], allow_none=True, + desc="has_presence_state", ) def get_presence_state(self, user_localpart): @@ -36,6 +38,7 @@ class PresenceStore(SQLBaseStore): table="presence", keyvalues={"user_id": user_localpart}, retcols=["state", "status_msg", "mtime"], + desc="get_presence_state", ) def set_presence_state(self, user_localpart, new_state): @@ -45,6 +48,7 @@ class PresenceStore(SQLBaseStore): updatevalues={"state": new_state["state"], "status_msg": new_state["status_msg"], "mtime": self._clock.time_msec()}, + desc="set_presence_state", ) def allow_presence_visible(self, observed_localpart, observer_userid): @@ -52,6 +56,7 @@ class PresenceStore(SQLBaseStore): table="presence_allow_inbound", values={"observed_user_id": observed_localpart, "observer_user_id": observer_userid}, + desc="allow_presence_visible", ) def disallow_presence_visible(self, observed_localpart, observer_userid): @@ -59,6 +64,7 @@ class PresenceStore(SQLBaseStore): table="presence_allow_inbound", keyvalues={"observed_user_id": observed_localpart, "observer_user_id": observer_userid}, + desc="disallow_presence_visible", ) def is_presence_visible(self, observed_localpart, observer_userid): @@ -68,6 +74,7 @@ class PresenceStore(SQLBaseStore): "observer_user_id": observer_userid}, retcols=["observed_user_id"], allow_none=True, + desc="is_presence_visible", ) def add_presence_list_pending(self, observer_localpart, observed_userid): @@ -76,6 +83,7 @@ class PresenceStore(SQLBaseStore): values={"user_id": observer_localpart, "observed_user_id": observed_userid, "accepted": False}, + desc="add_presence_list_pending", ) def set_presence_list_accepted(self, observer_localpart, observed_userid): @@ -84,6 +92,7 @@ class PresenceStore(SQLBaseStore): keyvalues={"user_id": observer_localpart, "observed_user_id": observed_userid}, updatevalues={"accepted": True}, + desc="set_presence_list_accepted", ) def get_presence_list(self, observer_localpart, accepted=None): @@ -95,6 +104,7 @@ class PresenceStore(SQLBaseStore): table="presence_list", keyvalues=keyvalues, retcols=["observed_user_id", "accepted"], + desc="get_presence_list", ) def del_presence_list(self, observer_localpart, observed_userid): @@ -102,4 +112,5 @@ class PresenceStore(SQLBaseStore): table="presence_list", keyvalues={"user_id": observer_localpart, "observed_user_id": observed_userid}, + desc="del_presence_list", ) diff --git a/synapse/storage/profile.py b/synapse/storage/profile.py index 153c7ad027..a6e52cb248 100644 --- a/synapse/storage/profile.py +++ b/synapse/storage/profile.py @@ -21,6 +21,7 @@ class ProfileStore(SQLBaseStore): return self._simple_insert( table="profiles", values={"user_id": user_localpart}, + desc="create_profile", ) def get_profile_displayname(self, user_localpart): @@ -28,6 +29,7 @@ class ProfileStore(SQLBaseStore): table="profiles", keyvalues={"user_id": user_localpart}, retcol="displayname", + desc="get_profile_displayname", ) def set_profile_displayname(self, user_localpart, new_displayname): @@ -35,6 +37,7 @@ class ProfileStore(SQLBaseStore): table="profiles", keyvalues={"user_id": user_localpart}, updatevalues={"displayname": new_displayname}, + desc="set_profile_displayname", ) def get_profile_avatar_url(self, user_localpart): @@ -42,6 +45,7 @@ class ProfileStore(SQLBaseStore): table="profiles", keyvalues={"user_id": user_localpart}, retcol="avatar_url", + desc="get_profile_avatar_url", ) def set_profile_avatar_url(self, user_localpart, new_avatar_url): @@ -49,4 +53,5 @@ class ProfileStore(SQLBaseStore): table="profiles", keyvalues={"user_id": user_localpart}, updatevalues={"avatar_url": new_avatar_url}, + desc="set_profile_avatar_url", ) diff --git a/synapse/storage/push_rule.py b/synapse/storage/push_rule.py index d769db2c78..c47bdc2861 100644 --- a/synapse/storage/push_rule.py +++ b/synapse/storage/push_rule.py @@ -50,7 +50,8 @@ class PushRuleStore(SQLBaseStore): results = yield self._simple_select_list( PushRuleEnableTable.table_name, {'user_name': user_name}, - PushRuleEnableTable.fields + PushRuleEnableTable.fields, + desc="get_push_rules_enabled_for_user", ) defer.returnValue( {r['rule_id']: False if r['enabled'] == 0 else True for r in results} @@ -201,7 +202,8 @@ class PushRuleStore(SQLBaseStore): """ yield self._simple_delete_one( PushRuleTable.table_name, - {'user_name': user_name, 'rule_id': rule_id} + {'user_name': user_name, 'rule_id': rule_id}, + desc="delete_push_rule", ) @defer.inlineCallbacks @@ -209,7 +211,8 @@ class PushRuleStore(SQLBaseStore): yield self._simple_upsert( PushRuleEnableTable.table_name, {'user_name': user_name, 'rule_id': rule_id}, - {'enabled': enabled} + {'enabled': enabled}, + desc="set_push_rule_enabled", ) diff --git a/synapse/storage/pusher.py b/synapse/storage/pusher.py index 587dada68f..000502b4ff 100644 --- a/synapse/storage/pusher.py +++ b/synapse/storage/pusher.py @@ -114,7 +114,9 @@ class PusherStore(SQLBaseStore): ts=pushkey_ts, lang=lang, data=data - )) + ), + desc="add_pusher", + ) except Exception as e: logger.error("create_pusher with failed: %s", e) raise StoreError(500, "Problem creating pusher.") @@ -123,7 +125,8 @@ class PusherStore(SQLBaseStore): def delete_pusher_by_app_id_pushkey(self, app_id, pushkey): yield self._simple_delete_one( PushersTable.table_name, - dict(app_id=app_id, pushkey=pushkey) + {"app_id": app_id, "pushkey": pushkey}, + desc="delete_pusher_by_app_id_pushkey", ) @defer.inlineCallbacks @@ -131,7 +134,8 @@ class PusherStore(SQLBaseStore): yield self._simple_update_one( PushersTable.table_name, {'app_id': app_id, 'pushkey': pushkey}, - {'last_token': last_token} + {'last_token': last_token}, + desc="update_pusher_last_token", ) @defer.inlineCallbacks @@ -140,7 +144,8 @@ class PusherStore(SQLBaseStore): yield self._simple_update_one( PushersTable.table_name, {'app_id': app_id, 'pushkey': pushkey}, - {'last_token': last_token, 'last_success': last_success} + {'last_token': last_token, 'last_success': last_success}, + desc="update_pusher_last_token_and_success", ) @defer.inlineCallbacks @@ -148,7 +153,8 @@ class PusherStore(SQLBaseStore): yield self._simple_update_one( PushersTable.table_name, {'app_id': app_id, 'pushkey': pushkey}, - {'failing_since': failing_since} + {'failing_since': failing_since}, + desc="update_pusher_failing_since", ) diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py index 3c2f1d6a15..f24154f146 100644 --- a/synapse/storage/registration.py +++ b/synapse/storage/registration.py @@ -39,7 +39,10 @@ class RegistrationStore(SQLBaseStore): Raises: StoreError if there was a problem adding this. """ - row = yield self._simple_select_one("users", {"name": user_id}, ["id"]) + row = yield self._simple_select_one( + "users", {"name": user_id}, ["id"], + desc="add_access_token_to_user", + ) if not row: raise StoreError(400, "Bad user ID supplied.") row_id = row["id"] @@ -48,7 +51,8 @@ class RegistrationStore(SQLBaseStore): { "user_id": row_id, "token": token - } + }, + desc="add_access_token_to_user", ) @defer.inlineCallbacks @@ -120,6 +124,7 @@ class RegistrationStore(SQLBaseStore): keyvalues={"name": user.to_string()}, retcol="admin", allow_none=True, + desc="is_server_admin", ) defer.returnValue(res if res else False) diff --git a/synapse/storage/rejections.py b/synapse/storage/rejections.py index 4e1a9a2783..0838eb3d12 100644 --- a/synapse/storage/rejections.py +++ b/synapse/storage/rejections.py @@ -29,7 +29,7 @@ class RejectionsStore(SQLBaseStore): "event_id": event_id, "reason": reason, "last_check": self._clock.time_msec(), - } + }, ) def get_rejection_reason(self, event_id): @@ -40,4 +40,5 @@ class RejectionsStore(SQLBaseStore): "event_id": event_id, }, allow_none=True, + desc="get_rejection_reason", ) diff --git a/synapse/storage/room.py b/synapse/storage/room.py index a16d321925..be3e28c2ea 100644 --- a/synapse/storage/room.py +++ b/synapse/storage/room.py @@ -52,7 +52,8 @@ class RoomStore(SQLBaseStore): "room_id": room_id, "creator": room_creator_user_id, "is_public": is_public, - } + }, + desc="store_room", ) except Exception as e: logger.error("store_room with room_id=%s failed: %s", room_id, e) @@ -70,6 +71,7 @@ class RoomStore(SQLBaseStore): table=RoomsTable.table_name, keyvalues={"room_id": room_id}, retcols=RoomsTable.fields, + desc="get_room", ) @defer.inlineCallbacks @@ -144,7 +146,7 @@ class RoomStore(SQLBaseStore): "event_id": event.event_id, "room_id": event.room_id, "topic": event.content["topic"], - } + }, ) def _store_room_name_txn(self, txn, event): diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py index 65ffb4627f..52c37c76f5 100644 --- a/synapse/storage/roommember.py +++ b/synapse/storage/roommember.py @@ -212,7 +212,8 @@ class RoomMemberStore(SQLBaseStore): return self._simple_select_onecol( "room_hosts", {"room_id": room_id}, - "host" + "host", + desc="get_joined_hosts_for_room", ) def _get_members_by_dict(self, where_dict): -- cgit 1.5.1 From 9a7f4962982e309877acb777eb358178579cacc9 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 23 Mar 2015 15:29:04 +0000 Subject: Sanitize RoomMemberStore --- synapse/handlers/room.py | 19 ------- synapse/storage/roommember.py | 124 +++++++++++++++++++++--------------------- 2 files changed, 61 insertions(+), 82 deletions(-) (limited to 'synapse/storage/roommember.py') diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index 823affc380..bc7f1c2402 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -310,25 +310,6 @@ class RoomMemberHandler(BaseHandler): # paginating defer.returnValue(chunk_data) - @defer.inlineCallbacks - def get_room_member(self, room_id, member_user_id, auth_user_id): - """Retrieve a room member from a room. - - Args: - room_id : The room the member is in. - member_user_id : The member's user ID - auth_user_id : The user ID of the user making this request. - Returns: - The room member, or None if this member does not exist. - Raises: - SynapseError if something goes wrong. - """ - yield self.auth.check_joined_room(room_id, auth_user_id) - - member = yield self.store.get_room_member(user_id=member_user_id, - room_id=room_id) - defer.returnValue(member) - @defer.inlineCallbacks def change_membership(self, event, context, do_auth=True): """ Change the membership status of a user in a room. diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py index 17ee4bb9ec..a229505208 100644 --- a/synapse/storage/roommember.py +++ b/synapse/storage/roommember.py @@ -74,23 +74,10 @@ class RoomMemberStore(SQLBaseStore): txn.execute(sql, (event.room_id, domain)) elif event.membership != Membership.INVITE: # Check if this was the last person to have left. - member_events = self._get_members_query_txn( - txn, - where_clause=("c.room_id = ? AND m.membership = ?" - " AND m.user_id != ?"), - where_values=(event.room_id, Membership.JOIN, target_user_id,) + joined_domains = self._get_joined_hosts_for_room_txn( + txn, event.room_id ) - joined_domains = set() - for e in member_events: - try: - joined_domains.add( - UserID.from_string(e.state_key).domain - ) - except: - # FIXME: How do we deal with invalid user ids in the db? - logger.exception("Invalid user_id: %s", event.state_key) - if domain not in joined_domains: sql = ( "DELETE FROM room_hosts WHERE room_id = ? AND host = ?" @@ -100,7 +87,6 @@ class RoomMemberStore(SQLBaseStore): self.get_rooms_for_user.invalidate(target_user_id) - @defer.inlineCallbacks def get_room_member(self, user_id, room_id): """Retrieve the current state of a room member. @@ -110,41 +96,27 @@ class RoomMemberStore(SQLBaseStore): Returns: Deferred: Results in a MembershipEvent or None. """ - rows = yield self._get_members_by_dict({ - "e.room_id": room_id, - "m.user_id": user_id, - }) + def f(txn): + events = self._get_members_events_txn( + txn, + room_id, + user_id=user_id, + ) - defer.returnValue(rows[0] if rows else None) + return events[0] if events else None - def _get_room_member(self, txn, user_id, room_id): - sql = ( - "SELECT e.* FROM events as e" - " INNER JOIN room_memberships as m" - " ON e.event_id = m.event_id" - " INNER JOIN current_state_events as c" - " ON m.event_id = c.event_id" - " WHERE m.user_id = ? and e.room_id = ?" - " LIMIT 1" - ) - txn.execute(sql, (user_id, room_id)) - rows = self.cursor_to_dict(txn) - if rows: - return self._parse_events_txn(txn, rows)[0] - else: - return None + return self.runInteraction("get_room_member", f) def get_users_in_room(self, room_id): def f(txn): - sql = ( - "SELECT m.user_id FROM room_memberships as m" - " INNER JOIN current_state_events as c" - " ON m.event_id = c.event_id" - " WHERE m.membership = ? AND m.room_id = ?" + + rows = self._get_members_rows_txn( + txn, + room_id=room_id, + membership=Membership.JOIN, ) - txn.execute(sql, (Membership.JOIN, room_id)) - return [r[0] for r in txn.fetchall()] + return [r["user_id"] for r in rows] return self.runInteraction("get_users_in_room", f) def get_room_members(self, room_id, membership=None): @@ -159,11 +131,14 @@ class RoomMemberStore(SQLBaseStore): list of namedtuples representing the members in this room. """ - where = {"m.room_id": room_id} - if membership: - where["m.membership"] = membership + def f(txn): + return self._get_members_events_txn( + txn, + room_id, + membership=membership, + ) - return self._get_members_by_dict(where) + return self.runInteraction("get_room_members", f) def get_rooms_for_user_where_membership_is(self, user_id, membership_list): """ Get all the rooms for this user where the membership for this user @@ -209,28 +184,52 @@ class RoomMemberStore(SQLBaseStore): ] def get_joined_hosts_for_room(self, room_id): - return self._simple_select_onecol( - "room_hosts", - {"room_id": room_id}, - "host", - desc="get_joined_hosts_for_room", + return self.runInteraction( + "get_joined_hosts_for_room", + self._get_joined_hosts_for_room_txn, + room_id, + ) + + def _get_joined_hosts_for_room_txn(self, txn, room_id): + rows = self._get_members_rows_txn( + txn, + room_id, membership=Membership.JOIN + ) + + joined_domains = set( + UserID.from_string(r["user_id"]).domain + for r in rows ) - def _get_members_by_dict(self, where_dict): - clause = " AND ".join("%s = ?" % k for k in where_dict.keys()) - vals = where_dict.values() - return self._get_members_query(clause, vals) + return joined_domains def _get_members_query(self, where_clause, where_values): return self.runInteraction( - "get_members_query", self._get_members_query_txn, + "get_members_query", self._get_members_events_txn, where_clause, where_values ) - def _get_members_query_txn(self, txn, where_clause, where_values): + def _get_members_events_txn(self, txn, room_id, membership=None, user_id=None): + rows = self._get_members_rows_txn( + txn, + room_id, membership, user_id, + ) + return self._get_events_txn(txn, [r["event_id"] for r in rows]) + + def _get_members_rows_txn(self, txn, room_id, membership=None, user_id=None): + where_clause = "c.room_id = ?" + where_values = [room_id] + + if membership: + where_clause += " AND m.membership = ?" + where_values.append(membership) + + if user_id: + where_clause += " AND m.user_id = ?" + where_values.append(user_id) + sql = ( - "SELECT e.* FROM events as e " - "INNER JOIN room_memberships as m " + "SELECT m.* FROM room_memberships as m " "ON e.event_id = m.event_id " "INNER JOIN current_state_events as c " "ON m.event_id = c.event_id " @@ -242,8 +241,7 @@ class RoomMemberStore(SQLBaseStore): txn.execute(sql, where_values) rows = self.cursor_to_dict(txn) - results = self._parse_events_txn(txn, rows) - return results + return rows @cached() def get_rooms_for_user(self, user_id): -- cgit 1.5.1 From 5a447098dd8442c7c69b2db22d42e6ce40c2dc6e Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 24 Mar 2015 16:19:24 +0000 Subject: Don't use room hosts table --- synapse/storage/roommember.py | 29 ++++------------------------- 1 file changed, 4 insertions(+), 25 deletions(-) (limited to 'synapse/storage/roommember.py') diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py index a229505208..df707f8124 100644 --- a/synapse/storage/roommember.py +++ b/synapse/storage/roommember.py @@ -65,26 +65,6 @@ class RoomMemberStore(SQLBaseStore): } ) - # Update room hosts table - if event.membership == Membership.JOIN: - sql = ( - "REPLACE INTO room_hosts (room_id, host) " - "VALUES (?, ?)" - ) - txn.execute(sql, (event.room_id, domain)) - elif event.membership != Membership.INVITE: - # Check if this was the last person to have left. - joined_domains = self._get_joined_hosts_for_room_txn( - txn, event.room_id - ) - - if domain not in joined_domains: - sql = ( - "DELETE FROM room_hosts WHERE room_id = ? AND host = ?" - ) - - txn.execute(sql, (event.room_id, domain)) - self.get_rooms_for_user.invalidate(target_user_id) def get_room_member(self, user_id, room_id): @@ -229,11 +209,10 @@ class RoomMemberStore(SQLBaseStore): where_values.append(user_id) sql = ( - "SELECT m.* FROM room_memberships as m " - "ON e.event_id = m.event_id " - "INNER JOIN current_state_events as c " - "ON m.event_id = c.event_id " - "WHERE %(where)s " + "SELECT m.* FROM room_memberships as m" + " INNER JOIN current_state_events as c" + " ON m.event_id = c.event_id" + " WHERE %(where)s" ) % { "where": where_clause, } -- cgit 1.5.1 From 0bfa78b39bf95ee24e78166c9545f59b34fd1d81 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 7 Apr 2015 12:16:05 +0100 Subject: PEP8 --- synapse/app/homeserver.py | 7 ++----- synapse/storage/__init__.py | 4 ---- synapse/storage/registration.py | 1 - synapse/storage/roommember.py | 1 - synapse/storage/state.py | 1 - synapse/storage/transactions.py | 1 - 6 files changed, 2 insertions(+), 13 deletions(-) (limited to 'synapse/storage/roommember.py') diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py index 95190c1d75..f7c724c4b4 100755 --- a/synapse/app/homeserver.py +++ b/synapse/app/homeserver.py @@ -17,9 +17,7 @@ import sys sys.dont_write_bytecode = True -from synapse.storage import ( - prepare_database, UpgradeDatabaseException, -) +from synapse.storage import UpgradeDatabaseException from synapse.storage.engines import create_engine from synapse.server import HomeServer @@ -60,7 +58,6 @@ import os import re import resource import subprocess -import sqlite3 import yaml @@ -322,7 +319,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) diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index e16414d736..231ec8169f 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -13,8 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -from twisted.internet import defer - from .appservice import ApplicationServiceStore from .directory import DirectoryStore from .events import EventsStore @@ -43,8 +41,6 @@ import logging import os import re -import threading - logger = logging.getLogger(__name__) diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py index b62b4a3414..0f9d898e5d 100644 --- a/synapse/storage/registration.py +++ b/synapse/storage/registration.py @@ -105,7 +105,6 @@ class RegistrationStore(SQLBaseStore): 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/roommember.py b/synapse/storage/roommember.py index df707f8124..8ea5756d61 100644 --- a/synapse/storage/roommember.py +++ b/synapse/storage/roommember.py @@ -40,7 +40,6 @@ class RoomMemberStore(SQLBaseStore): """ try: target_user_id = event.state_key - domain = UserID.from_string(target_user_id).domain except: logger.exception( "Failed to parse target_user_id=%s", target_user_id diff --git a/synapse/storage/state.py b/synapse/storage/state.py index 3e55cb81bf..4994bacd6c 100644 --- a/synapse/storage/state.py +++ b/synapse/storage/state.py @@ -159,4 +159,3 @@ class StateStore(SQLBaseStore): def _make_group_id(clock): return str(int(clock.time_msec())) + random_string(5) - diff --git a/synapse/storage/transactions.py b/synapse/storage/transactions.py index 9594fe1f2b..4c3dc58662 100644 --- a/synapse/storage/transactions.py +++ b/synapse/storage/transactions.py @@ -375,4 +375,3 @@ class DestinationsTable(object): "retry_last_ts", "retry_interval", ] - -- cgit 1.5.1 From 69d40636517d0634e6e4c598d6897bf367b1d634 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 30 Apr 2015 16:47:38 +0100 Subject: Add get_rooms_for_user cache --- synapse/storage/roommember.py | 2 ++ 1 file changed, 2 insertions(+) (limited to 'synapse/storage/roommember.py') diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py index 8ea5756d61..831169e220 100644 --- a/synapse/storage/roommember.py +++ b/synapse/storage/roommember.py @@ -65,6 +65,7 @@ class RoomMemberStore(SQLBaseStore): ) self.get_rooms_for_user.invalidate(target_user_id) + self.get_joined_hosts_for_room.invalidate(event.room_id) def get_room_member(self, user_id, room_id): """Retrieve the current state of a room member. @@ -162,6 +163,7 @@ class RoomMemberStore(SQLBaseStore): RoomsForUser(**r) for r in self.cursor_to_dict(txn) ] + @cached() def get_joined_hosts_for_room(self, room_id): return self.runInteraction( "get_joined_hosts_for_room", -- cgit 1.5.1 From 6ead27dddab13df042d8da4bbf64a10068358c3a Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 30 Apr 2015 18:32:03 +0100 Subject: Add more conditions on JOINs to make postgres go a little faster. --- synapse/storage/roommember.py | 2 +- synapse/storage/stream.py | 8 +++++--- 2 files changed, 6 insertions(+), 4 deletions(-) (limited to 'synapse/storage/roommember.py') diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py index 831169e220..2952c62f20 100644 --- a/synapse/storage/roommember.py +++ b/synapse/storage/roommember.py @@ -212,7 +212,7 @@ class RoomMemberStore(SQLBaseStore): sql = ( "SELECT m.* FROM room_memberships as m" " INNER JOIN current_state_events as c" - " ON m.event_id = c.event_id" + " ON m.event_id = c.event_id AND m.room_id = c.room_id " " WHERE %(where)s" ) % { "where": where_clause, diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index df6de7cbcd..280d4ad605 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -149,7 +149,8 @@ class StreamStore(SQLBaseStore): # select all the events between from/to with a sensible limit sql = ( "SELECT e.event_id, e.room_id, e.type, s.state_key, " - "e.stream_ordering FROM events AS e LEFT JOIN state_events as s ON " + "e.stream_ordering FROM events AS e " + "LEFT JOIN state_events as s ON " "e.event_id = s.event_id " "WHERE e.stream_ordering > ? AND e.stream_ordering <= ? " "ORDER BY stream_ordering ASC LIMIT %(limit)d " @@ -214,8 +215,9 @@ class StreamStore(SQLBaseStore): current_room_membership_sql = ( "SELECT m.room_id FROM room_memberships as m " - "INNER JOIN current_state_events as c ON m.event_id = c.event_id " - "WHERE m.user_id = ? AND m.membership = 'join'" + " INNER JOIN current_state_events as c" + " ON m.event_id = c.event_id AND c.state_key = m.user_id" + " WHERE m.user_id = ? AND m.membership = 'join'" ) # We also want to get any membership events about that user, e.g. -- cgit 1.5.1 From 61d05daab1ab4e00f40195cd0c81a88cdedb9632 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 30 Apr 2015 18:41:05 +0100 Subject: More join conditions --- synapse/storage/roommember.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) (limited to 'synapse/storage/roommember.py') diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py index 2952c62f20..09fb77a194 100644 --- a/synapse/storage/roommember.py +++ b/synapse/storage/roommember.py @@ -154,7 +154,9 @@ class RoomMemberStore(SQLBaseStore): "SELECT m.room_id, m.sender, m.membership" " FROM room_memberships as m" " INNER JOIN current_state_events as c" - " ON m.event_id = c.event_id" + " ON m.event_id = c.event_id " + " AND m.room_id = c.room_id " + " AND m.user_id = c.state_key" " WHERE %s" ) % (where_clause,) @@ -212,7 +214,9 @@ class RoomMemberStore(SQLBaseStore): sql = ( "SELECT m.* FROM room_memberships as m" " INNER JOIN current_state_events as c" - " ON m.event_id = c.event_id AND m.room_id = c.room_id " + " ON m.event_id = c.event_id " + " AND m.room_id = c.room_id " + " AND m.user_id = c.state_key" " WHERE %(where)s" ) % { "where": where_clause, -- cgit 1.5.1 From a9aea68fd568182185e8d0ae478c56df8ac6be49 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Tue, 5 May 2015 14:57:08 +0100 Subject: Invalidate the caches from the correct thread --- synapse/storage/event_federation.py | 10 ++++++---- synapse/storage/events.py | 39 ++++++++++++++++++++++++------------- synapse/storage/room.py | 4 ++-- synapse/storage/roommember.py | 8 +++++--- synapse/storage/signatures.py | 12 ++++++------ synapse/storage/state.py | 2 +- 6 files changed, 46 insertions(+), 29 deletions(-) (limited to 'synapse/storage/roommember.py') diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 68f39bd684..3cd3fbdc9b 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -241,7 +241,7 @@ class EventFederationStore(SQLBaseStore): return int(min_depth) if min_depth is not None else None - def _update_min_depth_for_room_txn(self, txn, room_id, depth): + def _update_min_depth_for_room_txn(self, txn, invalidates, room_id, depth): min_depth = self._get_min_depth_interaction(txn, room_id) do_insert = depth < min_depth if min_depth else True @@ -256,8 +256,8 @@ class EventFederationStore(SQLBaseStore): }, ) - def _handle_prev_events(self, txn, outlier, event_id, prev_events, - room_id): + def _handle_prev_events(self, txn, invalidates, outlier, event_id, + prev_events, room_id): """ For the given event, update the event edges table and forward and backward extremities tables. @@ -330,7 +330,9 @@ class EventFederationStore(SQLBaseStore): ) txn.execute(query) - self.get_latest_event_ids_in_room.invalidate(room_id) + invalidates.append(( + self.get_latest_event_ids_in_room.invalidate, room_id + )) def get_backfill_events(self, room_id, event_list, limit): """Get a list of Events for a given topic that occurred before (and diff --git a/synapse/storage/events.py b/synapse/storage/events.py index a3c260ddc4..b2ab4b02f3 100644 --- a/synapse/storage/events.py +++ b/synapse/storage/events.py @@ -42,7 +42,7 @@ class EventsStore(SQLBaseStore): stream_ordering = self.min_token try: - yield self.runInteraction( + invalidates = yield self.runInteraction( "persist_event", self._persist_event_txn, event=event, @@ -52,6 +52,11 @@ class EventsStore(SQLBaseStore): is_new_state=is_new_state, current_state=current_state, ) + for invalidated in invalidates: + invalidated_callback = invalidated[0] + invalidated_args = invalidated[1:] + invalidated_callback(*invalidated_args) + except _RollbackButIsFineException: pass @@ -91,9 +96,10 @@ class EventsStore(SQLBaseStore): def _persist_event_txn(self, txn, event, context, backfilled, stream_ordering=None, is_new_state=True, current_state=None): + invalidates = [] # Remove the any existing cache entries for the event_id - self._invalidate_get_event_cache(event.event_id) + invalidates.append((self._invalidate_get_event_cache, event.event_id)) if stream_ordering is None: with self._stream_id_gen.get_next_txn(txn) as stream_ordering: @@ -150,10 +156,11 @@ class EventsStore(SQLBaseStore): outlier = event.internal_metadata.is_outlier() if not outlier: - self._store_state_groups_txn(txn, event, context) + self._store_state_groups_txn(txn, invalidates, event, context) self._update_min_depth_for_room_txn( txn, + invalidates, event.room_id, event.depth ) @@ -199,6 +206,7 @@ class EventsStore(SQLBaseStore): self._handle_prev_events( txn, + invalidates, outlier=outlier, event_id=event.event_id, prev_events=event.prev_events, @@ -206,13 +214,13 @@ class EventsStore(SQLBaseStore): ) if event.type == EventTypes.Member: - self._store_room_member_txn(txn, event) + self._store_room_member_txn(txn, invalidates, event) elif event.type == EventTypes.Name: - self._store_room_name_txn(txn, event) + self._store_room_name_txn(txn, invalidates, event) elif event.type == EventTypes.Topic: - self._store_room_topic_txn(txn, event) + self._store_room_topic_txn(txn, invalidates, event) elif event.type == EventTypes.Redaction: - self._store_redaction(txn, event) + self._store_redaction(txn, invalidates, event) event_dict = { k: v @@ -281,19 +289,22 @@ class EventsStore(SQLBaseStore): ) if context.rejected: - self._store_rejections_txn(txn, event.event_id, context.rejected) + self._store_rejections_txn( + txn, invalidates, 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, + txn, invalidates, 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 + txn, invalidates, event.event_id, prev_event_id, alg, + hash_bytes ) for auth_id, _ in event.auth_events: @@ -309,7 +320,7 @@ class EventsStore(SQLBaseStore): (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 + txn, invalidates, event.event_id, ref_alg, ref_hash_bytes ) if event.is_state(): @@ -356,9 +367,11 @@ class EventsStore(SQLBaseStore): } ) - def _store_redaction(self, txn, event): + return invalidates + + def _store_redaction(self, txn, invalidates, event): # invalidate the cache for the redacted event - self._invalidate_get_event_cache(event.redacts) + invalidates.append((self._invalidate_get_event_cache, event.redacts)) txn.execute( "INSERT INTO redactions (event_id, redacts) VALUES (?,?)", (event.event_id, event.redacts) diff --git a/synapse/storage/room.py b/synapse/storage/room.py index f956377632..d42d7ff0e3 100644 --- a/synapse/storage/room.py +++ b/synapse/storage/room.py @@ -162,7 +162,7 @@ class RoomStore(SQLBaseStore): defer.returnValue(ret) - def _store_room_topic_txn(self, txn, event): + def _store_room_topic_txn(self, txn, invalidates, event): if hasattr(event, "content") and "topic" in event.content: self._simple_insert_txn( txn, @@ -174,7 +174,7 @@ class RoomStore(SQLBaseStore): }, ) - def _store_room_name_txn(self, txn, event): + def _store_room_name_txn(self, txn, invalidates, event): if hasattr(event, "content") and "name" in event.content: self._simple_insert_txn( txn, diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py index 09fb77a194..117da817ba 100644 --- a/synapse/storage/roommember.py +++ b/synapse/storage/roommember.py @@ -35,7 +35,7 @@ RoomsForUser = namedtuple( class RoomMemberStore(SQLBaseStore): - def _store_room_member_txn(self, txn, event): + def _store_room_member_txn(self, txn, invalidates, event): """Store a room member in the database. """ try: @@ -64,8 +64,10 @@ class RoomMemberStore(SQLBaseStore): } ) - self.get_rooms_for_user.invalidate(target_user_id) - self.get_joined_hosts_for_room.invalidate(event.room_id) + invalidates.extend([ + (self.get_rooms_for_user.invalidate, target_user_id), + (self.get_joined_hosts_for_room.invalidate, event.room_id), + ]) def get_room_member(self, user_id, room_id): """Retrieve the current state of a room member. diff --git a/synapse/storage/signatures.py b/synapse/storage/signatures.py index f051828630..e3979846e7 100644 --- a/synapse/storage/signatures.py +++ b/synapse/storage/signatures.py @@ -39,8 +39,8 @@ class SignatureStore(SQLBaseStore): txn.execute(query, (event_id, )) return dict(txn.fetchall()) - def _store_event_content_hash_txn(self, txn, event_id, algorithm, - hash_bytes): + def _store_event_content_hash_txn(self, txn, invalidates, event_id, + algorithm, hash_bytes): """Store a hash for a Event Args: txn (cursor): @@ -101,8 +101,8 @@ class SignatureStore(SQLBaseStore): txn.execute(query, (event_id, )) return {k: v for k, v in txn.fetchall()} - def _store_event_reference_hash_txn(self, txn, event_id, algorithm, - hash_bytes): + def _store_event_reference_hash_txn(self, txn, invalidates, event_id, + algorithm, hash_bytes): """Store a hash for a PDU Args: txn (cursor): @@ -184,8 +184,8 @@ class SignatureStore(SQLBaseStore): hashes[algorithm] = hash_bytes return results - def _store_prev_event_hash_txn(self, txn, event_id, prev_event_id, - algorithm, hash_bytes): + def _store_prev_event_hash_txn(self, txn, invalidates, event_id, + prev_event_id, algorithm, hash_bytes): self._simple_insert_txn( txn, "event_edge_hashes", diff --git a/synapse/storage/state.py b/synapse/storage/state.py index 7e55e8bed6..35d11c27cc 100644 --- a/synapse/storage/state.py +++ b/synapse/storage/state.py @@ -82,7 +82,7 @@ class StateStore(SQLBaseStore): f, ) - def _store_state_groups_txn(self, txn, event, context): + def _store_state_groups_txn(self, txn, invalidates, event, context): if context.current_state is None: return -- cgit 1.5.1 From d18f37e026a02b4e899bc96e600850007a613189 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Tue, 5 May 2015 17:32:21 +0100 Subject: Collect the invalidate callbacks on the transaction object rather than passing around a separate list --- synapse/storage/_base.py | 18 ++++++++++---- synapse/storage/event_federation.py | 10 ++++---- synapse/storage/events.py | 48 ++++++++++++++++--------------------- synapse/storage/room.py | 4 ++-- synapse/storage/roommember.py | 8 +++---- synapse/storage/signatures.py | 12 +++++----- synapse/storage/state.py | 2 +- 7 files changed, 51 insertions(+), 51 deletions(-) (limited to 'synapse/storage/roommember.py') diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 579ed56377..ccf9697fa3 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -185,12 +185,16 @@ class LoggingTransaction(object): """An object that almost-transparently proxies for the 'txn' object passed to the constructor. Adds logging and metrics to the .execute() method.""" - __slots__ = ["txn", "name", "database_engine"] + __slots__ = ["txn", "name", "database_engine", "after_callbacks"] - def __init__(self, txn, name, database_engine): + def __init__(self, txn, name, database_engine, after_callbacks): object.__setattr__(self, "txn", txn) object.__setattr__(self, "name", name) object.__setattr__(self, "database_engine", database_engine) + object.__setattr__(self, "after_callbacks", after_callbacks) + + def call_after(self, callback, *args): + self.after_callbacks.append((callback, args)) def __getattr__(self, name): return getattr(self.txn, name) @@ -336,6 +340,8 @@ class SQLBaseStore(object): start_time = time.time() * 1000 + after_callbacks = [] + def inner_func(conn, *args, **kwargs): with LoggingContext("runInteraction") as context: if self.database_engine.is_connection_closed(conn): @@ -360,10 +366,10 @@ class SQLBaseStore(object): while True: try: txn = conn.cursor() - return func( - LoggingTransaction(txn, name, self.database_engine), - *args, **kwargs + txn = LoggingTransaction( + txn, name, self.database_engine, after_callbacks ) + return func(txn, *args, **kwargs) except self.database_engine.module.OperationalError as e: # This can happen if the database disappears mid # transaction. @@ -412,6 +418,8 @@ class SQLBaseStore(object): result = yield self._db_pool.runWithConnection( inner_func, *args, **kwargs ) + for after_callback, after_args in after_callbacks: + after_callback(*after_args) defer.returnValue(result) def cursor_to_dict(self, cursor): diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 3cd3fbdc9b..893344eff3 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -241,7 +241,7 @@ class EventFederationStore(SQLBaseStore): return int(min_depth) if min_depth is not None else None - def _update_min_depth_for_room_txn(self, txn, invalidates, room_id, depth): + def _update_min_depth_for_room_txn(self, txn, room_id, depth): min_depth = self._get_min_depth_interaction(txn, room_id) do_insert = depth < min_depth if min_depth else True @@ -256,8 +256,8 @@ class EventFederationStore(SQLBaseStore): }, ) - def _handle_prev_events(self, txn, invalidates, outlier, event_id, - prev_events, room_id): + def _handle_prev_events(self, txn, outlier, event_id, prev_events, + room_id): """ For the given event, update the event edges table and forward and backward extremities tables. @@ -330,9 +330,9 @@ class EventFederationStore(SQLBaseStore): ) txn.execute(query) - invalidates.append(( + txn.call_after( self.get_latest_event_ids_in_room.invalidate, room_id - )) + ) def get_backfill_events(self, room_id, event_list, limit): """Get a list of Events for a given topic that occurred before (and diff --git a/synapse/storage/events.py b/synapse/storage/events.py index 7dc49ceed6..17f9d27289 100644 --- a/synapse/storage/events.py +++ b/synapse/storage/events.py @@ -42,7 +42,7 @@ class EventsStore(SQLBaseStore): stream_ordering = self.min_token try: - invalidates = yield self.runInteraction( + yield self.runInteraction( "persist_event", self._persist_event_txn, event=event, @@ -52,11 +52,6 @@ class EventsStore(SQLBaseStore): is_new_state=is_new_state, current_state=current_state, ) - for invalidated in invalidates: - invalidated_callback = invalidated[0] - invalidated_args = invalidated[1:] - invalidated_callback(*invalidated_args) - except _RollbackButIsFineException: pass @@ -96,10 +91,9 @@ class EventsStore(SQLBaseStore): def _persist_event_txn(self, txn, event, context, backfilled, stream_ordering=None, is_new_state=True, current_state=None): - invalidates = [] # Remove the any existing cache entries for the event_id - invalidates.append((self._invalidate_get_event_cache, event.event_id)) + txn.call_after(self._invalidate_get_event_cache, event.event_id) if stream_ordering is None: with self._stream_id_gen.get_next_txn(txn) as stream_ordering: @@ -121,10 +115,12 @@ class EventsStore(SQLBaseStore): for s in current_state: if s.type == EventTypes.Member: - invalidates.extend([ - (self.get_rooms_for_user.invalidate, s.state_key), - (self.get_joined_hosts_for_room.invalidate, s.room_id), - ]) + txn.call_after( + self.get_rooms_for_user.invalidate, s.state_key + ) + txn.call_after( + self.get_joined_hosts_for_room.invalidate, s.room_id + ) self._simple_insert_txn( txn, "current_state_events", @@ -161,11 +157,10 @@ class EventsStore(SQLBaseStore): outlier = event.internal_metadata.is_outlier() if not outlier: - self._store_state_groups_txn(txn, invalidates, event, context) + self._store_state_groups_txn(txn, event, context) self._update_min_depth_for_room_txn( txn, - invalidates, event.room_id, event.depth ) @@ -207,11 +202,10 @@ class EventsStore(SQLBaseStore): sql, (False, event.event_id,) ) - return invalidates + return self._handle_prev_events( txn, - invalidates, outlier=outlier, event_id=event.event_id, prev_events=event.prev_events, @@ -219,13 +213,13 @@ class EventsStore(SQLBaseStore): ) if event.type == EventTypes.Member: - self._store_room_member_txn(txn, invalidates, event) + self._store_room_member_txn(txn, event) elif event.type == EventTypes.Name: - self._store_room_name_txn(txn, invalidates, event) + self._store_room_name_txn(txn, event) elif event.type == EventTypes.Topic: - self._store_room_topic_txn(txn, invalidates, event) + self._store_room_topic_txn(txn, event) elif event.type == EventTypes.Redaction: - self._store_redaction(txn, invalidates, event) + self._store_redaction(txn, event) event_dict = { k: v @@ -295,20 +289,20 @@ class EventsStore(SQLBaseStore): if context.rejected: self._store_rejections_txn( - txn, invalidates, event.event_id, context.rejected + 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, invalidates, event.event_id, hash_alg, hash_bytes, + 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, invalidates, event.event_id, prev_event_id, alg, + txn, event.event_id, prev_event_id, alg, hash_bytes ) @@ -325,7 +319,7 @@ class EventsStore(SQLBaseStore): (ref_alg, ref_hash_bytes) = compute_event_reference_hash(event) self._store_event_reference_hash_txn( - txn, invalidates, event.event_id, ref_alg, ref_hash_bytes + txn, event.event_id, ref_alg, ref_hash_bytes ) if event.is_state(): @@ -372,11 +366,11 @@ class EventsStore(SQLBaseStore): } ) - return invalidates + return - def _store_redaction(self, txn, invalidates, event): + def _store_redaction(self, txn, event): # invalidate the cache for the redacted event - invalidates.append((self._invalidate_get_event_cache, event.redacts)) + txn.call_after(self._invalidate_get_event_cache, event.redacts) txn.execute( "INSERT INTO redactions (event_id, redacts) VALUES (?,?)", (event.event_id, event.redacts) diff --git a/synapse/storage/room.py b/synapse/storage/room.py index d42d7ff0e3..f956377632 100644 --- a/synapse/storage/room.py +++ b/synapse/storage/room.py @@ -162,7 +162,7 @@ class RoomStore(SQLBaseStore): defer.returnValue(ret) - def _store_room_topic_txn(self, txn, invalidates, event): + def _store_room_topic_txn(self, txn, event): if hasattr(event, "content") and "topic" in event.content: self._simple_insert_txn( txn, @@ -174,7 +174,7 @@ class RoomStore(SQLBaseStore): }, ) - def _store_room_name_txn(self, txn, invalidates, event): + def _store_room_name_txn(self, txn, event): if hasattr(event, "content") and "name" in event.content: self._simple_insert_txn( txn, diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py index 117da817ba..839c74f63a 100644 --- a/synapse/storage/roommember.py +++ b/synapse/storage/roommember.py @@ -35,7 +35,7 @@ RoomsForUser = namedtuple( class RoomMemberStore(SQLBaseStore): - def _store_room_member_txn(self, txn, invalidates, event): + def _store_room_member_txn(self, txn, event): """Store a room member in the database. """ try: @@ -64,10 +64,8 @@ class RoomMemberStore(SQLBaseStore): } ) - invalidates.extend([ - (self.get_rooms_for_user.invalidate, target_user_id), - (self.get_joined_hosts_for_room.invalidate, event.room_id), - ]) + txn.call_after(self.get_rooms_for_user.invalidate, target_user_id) + txn.call_after(self.get_joined_hosts_for_room.invalidate, event.room_id) def get_room_member(self, user_id, room_id): """Retrieve the current state of a room member. diff --git a/synapse/storage/signatures.py b/synapse/storage/signatures.py index e3979846e7..f051828630 100644 --- a/synapse/storage/signatures.py +++ b/synapse/storage/signatures.py @@ -39,8 +39,8 @@ class SignatureStore(SQLBaseStore): txn.execute(query, (event_id, )) return dict(txn.fetchall()) - def _store_event_content_hash_txn(self, txn, invalidates, event_id, - algorithm, hash_bytes): + def _store_event_content_hash_txn(self, txn, event_id, algorithm, + hash_bytes): """Store a hash for a Event Args: txn (cursor): @@ -101,8 +101,8 @@ class SignatureStore(SQLBaseStore): txn.execute(query, (event_id, )) return {k: v for k, v in txn.fetchall()} - def _store_event_reference_hash_txn(self, txn, invalidates, event_id, - algorithm, hash_bytes): + def _store_event_reference_hash_txn(self, txn, event_id, algorithm, + hash_bytes): """Store a hash for a PDU Args: txn (cursor): @@ -184,8 +184,8 @@ class SignatureStore(SQLBaseStore): hashes[algorithm] = hash_bytes return results - def _store_prev_event_hash_txn(self, txn, invalidates, event_id, - prev_event_id, algorithm, hash_bytes): + def _store_prev_event_hash_txn(self, txn, event_id, prev_event_id, + algorithm, hash_bytes): self._simple_insert_txn( txn, "event_edge_hashes", diff --git a/synapse/storage/state.py b/synapse/storage/state.py index 35d11c27cc..7e55e8bed6 100644 --- a/synapse/storage/state.py +++ b/synapse/storage/state.py @@ -82,7 +82,7 @@ class StateStore(SQLBaseStore): f, ) - def _store_state_groups_txn(self, txn, invalidates, event, context): + def _store_state_groups_txn(self, txn, event, context): if context.current_state is None: return -- cgit 1.5.1