mirror of
https://git.anonymousland.org/anonymousland/synapse.git
synced 2025-05-02 11:06:07 -04:00
Split out SignatureStore and EventFederationStore
This commit is contained in:
Erik Johnston
parent
17445e6701
commit
a9a2d66cdd
3 changed files with 148 additions and 174 deletions
|
|
@ -15,7 +15,10 @@
|
|||
|
||||
from twisted.internet import defer
|
||||
|
||||
from ._base import SQLBaseStore
|
||||
from synapse.storage._base import SQLBaseStore
|
||||
from synapse.storage.events import EventsWorkerStore
|
||||
from synapse.storage.signatures import SignatureWorkerStore
|
||||
|
||||
from synapse.api.errors import StoreError
|
||||
from synapse.util.caches.descriptors import cached
|
||||
from unpaddedbase64 import encode_base64
|
||||
|
|
@ -27,30 +30,8 @@ from Queue import PriorityQueue, Empty
|
|||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class EventFederationStore(SQLBaseStore):
|
||||
""" Responsible for storing and serving up the various graphs associated
|
||||
with an event. Including the main event graph and the auth chains for an
|
||||
event.
|
||||
|
||||
Also has methods for getting the front (latest) and back (oldest) edges
|
||||
of the event graphs. These are used to generate the parents for new events
|
||||
and backfilling from another server respectively.
|
||||
"""
|
||||
|
||||
EVENT_AUTH_STATE_ONLY = "event_auth_state_only"
|
||||
|
||||
def __init__(self, db_conn, hs):
|
||||
super(EventFederationStore, self).__init__(db_conn, hs)
|
||||
|
||||
self.register_background_update_handler(
|
||||
self.EVENT_AUTH_STATE_ONLY,
|
||||
self._background_delete_non_state_event_auth,
|
||||
)
|
||||
|
||||
hs.get_clock().looping_call(
|
||||
self._delete_old_forward_extrem_cache, 60 * 60 * 1000
|
||||
)
|
||||
|
||||
class EventFederationWorkerStore(EventsWorkerStore, SignatureWorkerStore,
|
||||
SQLBaseStore):
|
||||
def get_auth_chain(self, event_ids, include_given=False):
|
||||
"""Get auth events for given event_ids. The events *must* be state events.
|
||||
|
||||
|
|
@ -228,88 +209,6 @@ 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):
|
||||
min_depth = self._get_min_depth_interaction(txn, room_id)
|
||||
|
||||
if min_depth and depth >= min_depth:
|
||||
return
|
||||
|
||||
self._simple_upsert_txn(
|
||||
txn,
|
||||
table="room_depth",
|
||||
keyvalues={
|
||||
"room_id": room_id,
|
||||
},
|
||||
values={
|
||||
"min_depth": depth,
|
||||
},
|
||||
)
|
||||
|
||||
def _handle_mult_prev_events(self, txn, events):
|
||||
"""
|
||||
For the given event, update the event edges table and forward and
|
||||
backward extremities tables.
|
||||
"""
|
||||
self._simple_insert_many_txn(
|
||||
txn,
|
||||
table="event_edges",
|
||||
values=[
|
||||
{
|
||||
"event_id": ev.event_id,
|
||||
"prev_event_id": e_id,
|
||||
"room_id": ev.room_id,
|
||||
"is_state": False,
|
||||
}
|
||||
for ev in events
|
||||
for e_id, _ in ev.prev_events
|
||||
],
|
||||
)
|
||||
|
||||
self._update_backward_extremeties(txn, events)
|
||||
|
||||
def _update_backward_extremeties(self, txn, events):
|
||||
"""Updates the event_backward_extremities tables based on the new/updated
|
||||
events being persisted.
|
||||
|
||||
This is called for new events *and* for events that were outliers, but
|
||||
are now being persisted as non-outliers.
|
||||
|
||||
Forward extremities are handled when we first start persisting the events.
|
||||
"""
|
||||
events_by_room = {}
|
||||
for ev in events:
|
||||
events_by_room.setdefault(ev.room_id, []).append(ev)
|
||||
|
||||
query = (
|
||||
"INSERT INTO event_backward_extremities (event_id, room_id)"
|
||||
" SELECT ?, ? WHERE NOT EXISTS ("
|
||||
" SELECT 1 FROM event_backward_extremities"
|
||||
" WHERE event_id = ? AND room_id = ?"
|
||||
" )"
|
||||
" AND NOT EXISTS ("
|
||||
" SELECT 1 FROM events WHERE event_id = ? AND room_id = ? "
|
||||
" AND outlier = ?"
|
||||
" )"
|
||||
)
|
||||
|
||||
txn.executemany(query, [
|
||||
(e_id, ev.room_id, e_id, ev.room_id, e_id, ev.room_id, False)
|
||||
for ev in events for e_id, _ in ev.prev_events
|
||||
if not ev.internal_metadata.is_outlier()
|
||||
])
|
||||
|
||||
query = (
|
||||
"DELETE FROM event_backward_extremities"
|
||||
" WHERE event_id = ? AND room_id = ?"
|
||||
)
|
||||
txn.executemany(
|
||||
query,
|
||||
[
|
||||
(ev.event_id, ev.room_id) for ev in events
|
||||
if not ev.internal_metadata.is_outlier()
|
||||
]
|
||||
)
|
||||
|
||||
def get_forward_extremeties_for_room(self, room_id, stream_ordering):
|
||||
"""For a given room_id and stream_ordering, return the forward
|
||||
extremeties of the room at that point in "time".
|
||||
|
|
@ -371,28 +270,6 @@ class EventFederationStore(SQLBaseStore):
|
|||
get_forward_extremeties_for_room_txn
|
||||
)
|
||||
|
||||
def _delete_old_forward_extrem_cache(self):
|
||||
def _delete_old_forward_extrem_cache_txn(txn):
|
||||
# Delete entries older than a month, while making sure we don't delete
|
||||
# the only entries for a room.
|
||||
sql = ("""
|
||||
DELETE FROM stream_ordering_to_exterm
|
||||
WHERE
|
||||
room_id IN (
|
||||
SELECT room_id
|
||||
FROM stream_ordering_to_exterm
|
||||
WHERE stream_ordering > ?
|
||||
) AND stream_ordering < ?
|
||||
""")
|
||||
txn.execute(
|
||||
sql,
|
||||
(self.stream_ordering_month_ago, self.stream_ordering_month_ago,)
|
||||
)
|
||||
return self.runInteraction(
|
||||
"_delete_old_forward_extrem_cache",
|
||||
_delete_old_forward_extrem_cache_txn
|
||||
)
|
||||
|
||||
def get_backfill_events(self, room_id, event_list, limit):
|
||||
"""Get a list of Events for a given topic that occurred before (and
|
||||
including) the events in event_list. Return a list of max size `limit`
|
||||
|
|
@ -522,6 +399,135 @@ class EventFederationStore(SQLBaseStore):
|
|||
|
||||
return event_results
|
||||
|
||||
|
||||
class EventFederationStore(EventFederationWorkerStore):
|
||||
""" Responsible for storing and serving up the various graphs associated
|
||||
with an event. Including the main event graph and the auth chains for an
|
||||
event.
|
||||
|
||||
Also has methods for getting the front (latest) and back (oldest) edges
|
||||
of the event graphs. These are used to generate the parents for new events
|
||||
and backfilling from another server respectively.
|
||||
"""
|
||||
|
||||
EVENT_AUTH_STATE_ONLY = "event_auth_state_only"
|
||||
|
||||
def __init__(self, db_conn, hs):
|
||||
super(EventFederationStore, self).__init__(db_conn, hs)
|
||||
|
||||
self.register_background_update_handler(
|
||||
self.EVENT_AUTH_STATE_ONLY,
|
||||
self._background_delete_non_state_event_auth,
|
||||
)
|
||||
|
||||
hs.get_clock().looping_call(
|
||||
self._delete_old_forward_extrem_cache, 60 * 60 * 1000
|
||||
)
|
||||
|
||||
def _update_min_depth_for_room_txn(self, txn, room_id, depth):
|
||||
min_depth = self._get_min_depth_interaction(txn, room_id)
|
||||
|
||||
if min_depth and depth >= min_depth:
|
||||
return
|
||||
|
||||
self._simple_upsert_txn(
|
||||
txn,
|
||||
table="room_depth",
|
||||
keyvalues={
|
||||
"room_id": room_id,
|
||||
},
|
||||
values={
|
||||
"min_depth": depth,
|
||||
},
|
||||
)
|
||||
|
||||
def _handle_mult_prev_events(self, txn, events):
|
||||
"""
|
||||
For the given event, update the event edges table and forward and
|
||||
backward extremities tables.
|
||||
"""
|
||||
self._simple_insert_many_txn(
|
||||
txn,
|
||||
table="event_edges",
|
||||
values=[
|
||||
{
|
||||
"event_id": ev.event_id,
|
||||
"prev_event_id": e_id,
|
||||
"room_id": ev.room_id,
|
||||
"is_state": False,
|
||||
}
|
||||
for ev in events
|
||||
for e_id, _ in ev.prev_events
|
||||
],
|
||||
)
|
||||
|
||||
self._update_backward_extremeties(txn, events)
|
||||
|
||||
def _update_backward_extremeties(self, txn, events):
|
||||
"""Updates the event_backward_extremities tables based on the new/updated
|
||||
events being persisted.
|
||||
|
||||
This is called for new events *and* for events that were outliers, but
|
||||
are now being persisted as non-outliers.
|
||||
|
||||
Forward extremities are handled when we first start persisting the events.
|
||||
"""
|
||||
events_by_room = {}
|
||||
for ev in events:
|
||||
events_by_room.setdefault(ev.room_id, []).append(ev)
|
||||
|
||||
query = (
|
||||
"INSERT INTO event_backward_extremities (event_id, room_id)"
|
||||
" SELECT ?, ? WHERE NOT EXISTS ("
|
||||
" SELECT 1 FROM event_backward_extremities"
|
||||
" WHERE event_id = ? AND room_id = ?"
|
||||
" )"
|
||||
" AND NOT EXISTS ("
|
||||
" SELECT 1 FROM events WHERE event_id = ? AND room_id = ? "
|
||||
" AND outlier = ?"
|
||||
" )"
|
||||
)
|
||||
|
||||
txn.executemany(query, [
|
||||
(e_id, ev.room_id, e_id, ev.room_id, e_id, ev.room_id, False)
|
||||
for ev in events for e_id, _ in ev.prev_events
|
||||
if not ev.internal_metadata.is_outlier()
|
||||
])
|
||||
|
||||
query = (
|
||||
"DELETE FROM event_backward_extremities"
|
||||
" WHERE event_id = ? AND room_id = ?"
|
||||
)
|
||||
txn.executemany(
|
||||
query,
|
||||
[
|
||||
(ev.event_id, ev.room_id) for ev in events
|
||||
if not ev.internal_metadata.is_outlier()
|
||||
]
|
||||
)
|
||||
|
||||
def _delete_old_forward_extrem_cache(self):
|
||||
def _delete_old_forward_extrem_cache_txn(txn):
|
||||
# Delete entries older than a month, while making sure we don't delete
|
||||
# the only entries for a room.
|
||||
sql = ("""
|
||||
DELETE FROM stream_ordering_to_exterm
|
||||
WHERE
|
||||
room_id IN (
|
||||
SELECT room_id
|
||||
FROM stream_ordering_to_exterm
|
||||
WHERE stream_ordering > ?
|
||||
) AND stream_ordering < ?
|
||||
""")
|
||||
txn.execute(
|
||||
sql,
|
||||
(self.stream_ordering_month_ago, self.stream_ordering_month_ago,)
|
||||
)
|
||||
return self.runInteraction(
|
||||
"_delete_old_forward_extrem_cache",
|
||||
_delete_old_forward_extrem_cache_txn
|
||||
)
|
||||
|
||||
def clean_room_for_join(self, room_id):
|
||||
return self.runInteraction(
|
||||
"clean_room_for_join",
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue