mirror of
https://github.com/element-hq/synapse
synced 2024-08-21 11:20:20 +00:00
Invalidate the caches from the correct thread
This commit is contained in:
parent
261d809a47
commit
a9aea68fd5
6 changed files with 46 additions and 29 deletions
|
@ -241,7 +241,7 @@ class EventFederationStore(SQLBaseStore):
|
||||||
|
|
||||||
return int(min_depth) if min_depth is not None else None
|
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)
|
min_depth = self._get_min_depth_interaction(txn, room_id)
|
||||||
|
|
||||||
do_insert = depth < min_depth if min_depth else True
|
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,
|
def _handle_prev_events(self, txn, invalidates, outlier, event_id,
|
||||||
room_id):
|
prev_events, room_id):
|
||||||
"""
|
"""
|
||||||
For the given event, update the event edges table and forward and
|
For the given event, update the event edges table and forward and
|
||||||
backward extremities tables.
|
backward extremities tables.
|
||||||
|
@ -330,7 +330,9 @@ class EventFederationStore(SQLBaseStore):
|
||||||
)
|
)
|
||||||
txn.execute(query)
|
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):
|
def get_backfill_events(self, room_id, event_list, limit):
|
||||||
"""Get a list of Events for a given topic that occurred before (and
|
"""Get a list of Events for a given topic that occurred before (and
|
||||||
|
|
|
@ -42,7 +42,7 @@ class EventsStore(SQLBaseStore):
|
||||||
stream_ordering = self.min_token
|
stream_ordering = self.min_token
|
||||||
|
|
||||||
try:
|
try:
|
||||||
yield self.runInteraction(
|
invalidates = yield self.runInteraction(
|
||||||
"persist_event",
|
"persist_event",
|
||||||
self._persist_event_txn,
|
self._persist_event_txn,
|
||||||
event=event,
|
event=event,
|
||||||
|
@ -52,6 +52,11 @@ class EventsStore(SQLBaseStore):
|
||||||
is_new_state=is_new_state,
|
is_new_state=is_new_state,
|
||||||
current_state=current_state,
|
current_state=current_state,
|
||||||
)
|
)
|
||||||
|
for invalidated in invalidates:
|
||||||
|
invalidated_callback = invalidated[0]
|
||||||
|
invalidated_args = invalidated[1:]
|
||||||
|
invalidated_callback(*invalidated_args)
|
||||||
|
|
||||||
except _RollbackButIsFineException:
|
except _RollbackButIsFineException:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
@ -91,9 +96,10 @@ class EventsStore(SQLBaseStore):
|
||||||
def _persist_event_txn(self, txn, event, context, backfilled,
|
def _persist_event_txn(self, txn, event, context, backfilled,
|
||||||
stream_ordering=None, is_new_state=True,
|
stream_ordering=None, is_new_state=True,
|
||||||
current_state=None):
|
current_state=None):
|
||||||
|
invalidates = []
|
||||||
|
|
||||||
# Remove the any existing cache entries for the event_id
|
# 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:
|
if stream_ordering is None:
|
||||||
with self._stream_id_gen.get_next_txn(txn) as stream_ordering:
|
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()
|
outlier = event.internal_metadata.is_outlier()
|
||||||
|
|
||||||
if not 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(
|
self._update_min_depth_for_room_txn(
|
||||||
txn,
|
txn,
|
||||||
|
invalidates,
|
||||||
event.room_id,
|
event.room_id,
|
||||||
event.depth
|
event.depth
|
||||||
)
|
)
|
||||||
|
@ -199,6 +206,7 @@ class EventsStore(SQLBaseStore):
|
||||||
|
|
||||||
self._handle_prev_events(
|
self._handle_prev_events(
|
||||||
txn,
|
txn,
|
||||||
|
invalidates,
|
||||||
outlier=outlier,
|
outlier=outlier,
|
||||||
event_id=event.event_id,
|
event_id=event.event_id,
|
||||||
prev_events=event.prev_events,
|
prev_events=event.prev_events,
|
||||||
|
@ -206,13 +214,13 @@ class EventsStore(SQLBaseStore):
|
||||||
)
|
)
|
||||||
|
|
||||||
if event.type == EventTypes.Member:
|
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:
|
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:
|
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:
|
elif event.type == EventTypes.Redaction:
|
||||||
self._store_redaction(txn, event)
|
self._store_redaction(txn, invalidates, event)
|
||||||
|
|
||||||
event_dict = {
|
event_dict = {
|
||||||
k: v
|
k: v
|
||||||
|
@ -281,19 +289,22 @@ class EventsStore(SQLBaseStore):
|
||||||
)
|
)
|
||||||
|
|
||||||
if context.rejected:
|
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():
|
for hash_alg, hash_base64 in event.hashes.items():
|
||||||
hash_bytes = decode_base64(hash_base64)
|
hash_bytes = decode_base64(hash_base64)
|
||||||
self._store_event_content_hash_txn(
|
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 prev_event_id, prev_hashes in event.prev_events:
|
||||||
for alg, hash_base64 in prev_hashes.items():
|
for alg, hash_base64 in prev_hashes.items():
|
||||||
hash_bytes = decode_base64(hash_base64)
|
hash_bytes = decode_base64(hash_base64)
|
||||||
self._store_prev_event_hash_txn(
|
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:
|
for auth_id, _ in event.auth_events:
|
||||||
|
@ -309,7 +320,7 @@ class EventsStore(SQLBaseStore):
|
||||||
|
|
||||||
(ref_alg, ref_hash_bytes) = compute_event_reference_hash(event)
|
(ref_alg, ref_hash_bytes) = compute_event_reference_hash(event)
|
||||||
self._store_event_reference_hash_txn(
|
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():
|
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
|
# 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(
|
txn.execute(
|
||||||
"INSERT INTO redactions (event_id, redacts) VALUES (?,?)",
|
"INSERT INTO redactions (event_id, redacts) VALUES (?,?)",
|
||||||
(event.event_id, event.redacts)
|
(event.event_id, event.redacts)
|
||||||
|
|
|
@ -162,7 +162,7 @@ class RoomStore(SQLBaseStore):
|
||||||
|
|
||||||
defer.returnValue(ret)
|
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:
|
if hasattr(event, "content") and "topic" in event.content:
|
||||||
self._simple_insert_txn(
|
self._simple_insert_txn(
|
||||||
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:
|
if hasattr(event, "content") and "name" in event.content:
|
||||||
self._simple_insert_txn(
|
self._simple_insert_txn(
|
||||||
txn,
|
txn,
|
||||||
|
|
|
@ -35,7 +35,7 @@ RoomsForUser = namedtuple(
|
||||||
|
|
||||||
class RoomMemberStore(SQLBaseStore):
|
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.
|
"""Store a room member in the database.
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
|
@ -64,8 +64,10 @@ class RoomMemberStore(SQLBaseStore):
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
|
|
||||||
self.get_rooms_for_user.invalidate(target_user_id)
|
invalidates.extend([
|
||||||
self.get_joined_hosts_for_room.invalidate(event.room_id)
|
(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):
|
def get_room_member(self, user_id, room_id):
|
||||||
"""Retrieve the current state of a room member.
|
"""Retrieve the current state of a room member.
|
||||||
|
|
|
@ -39,8 +39,8 @@ class SignatureStore(SQLBaseStore):
|
||||||
txn.execute(query, (event_id, ))
|
txn.execute(query, (event_id, ))
|
||||||
return dict(txn.fetchall())
|
return dict(txn.fetchall())
|
||||||
|
|
||||||
def _store_event_content_hash_txn(self, txn, event_id, algorithm,
|
def _store_event_content_hash_txn(self, txn, invalidates, event_id,
|
||||||
hash_bytes):
|
algorithm, hash_bytes):
|
||||||
"""Store a hash for a Event
|
"""Store a hash for a Event
|
||||||
Args:
|
Args:
|
||||||
txn (cursor):
|
txn (cursor):
|
||||||
|
@ -101,8 +101,8 @@ class SignatureStore(SQLBaseStore):
|
||||||
txn.execute(query, (event_id, ))
|
txn.execute(query, (event_id, ))
|
||||||
return {k: v for k, v in txn.fetchall()}
|
return {k: v for k, v in txn.fetchall()}
|
||||||
|
|
||||||
def _store_event_reference_hash_txn(self, txn, event_id, algorithm,
|
def _store_event_reference_hash_txn(self, txn, invalidates, event_id,
|
||||||
hash_bytes):
|
algorithm, hash_bytes):
|
||||||
"""Store a hash for a PDU
|
"""Store a hash for a PDU
|
||||||
Args:
|
Args:
|
||||||
txn (cursor):
|
txn (cursor):
|
||||||
|
@ -184,8 +184,8 @@ class SignatureStore(SQLBaseStore):
|
||||||
hashes[algorithm] = hash_bytes
|
hashes[algorithm] = hash_bytes
|
||||||
return results
|
return results
|
||||||
|
|
||||||
def _store_prev_event_hash_txn(self, txn, event_id, prev_event_id,
|
def _store_prev_event_hash_txn(self, txn, invalidates, event_id,
|
||||||
algorithm, hash_bytes):
|
prev_event_id, algorithm, hash_bytes):
|
||||||
self._simple_insert_txn(
|
self._simple_insert_txn(
|
||||||
txn,
|
txn,
|
||||||
"event_edge_hashes",
|
"event_edge_hashes",
|
||||||
|
|
|
@ -82,7 +82,7 @@ class StateStore(SQLBaseStore):
|
||||||
f,
|
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:
|
if context.current_state is None:
|
||||||
return
|
return
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue