dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-08-14 11:30:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Force event_edges_drop_invalid_rows update into foreground

Browse Source
This commit is contained in:
Olivier 'reivilibre
2026-08-06 12:12:25 +01:00
parent b4eac51467
commit e64d9f5c15
3 changed files with 74 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/storage/databases/main/events_bg_updates.py
-101
View File
@@ -280,11 +280,6 @@ class EventsBackgroundUpdatesStore(
################################################################################
self.db_pool.updates.register_background_update_handler(
_BackgroundUpdates.EVENT_EDGES_DROP_INVALID_ROWS,
self._background_drop_invalid_event_edges_rows,
)
self.db_pool.updates.register_background_index_update(
_BackgroundUpdates.EVENT_EDGES_REPLACE_INDEX,
index_name="event_edges_event_id_prev_event_id_idx",
@@ -1497,102 +1492,6 @@ class EventsBackgroundUpdatesStore(
return 0
async def _background_drop_invalid_event_edges_rows(
self, progress: JsonDict, batch_size: int
) -> int:
"""Drop invalid rows from event_edges
This only runs for postgres. For SQLite, it all happens synchronously.
Firstly, drop any rows with is_state=True. These may have been added a long time
ago, but they are no longer used.
We also drop rows that do not correspond to entries in `events`, and add a
foreign key.
"""
last_event_id = progress.get("last_event_id", "")
def drop_invalid_event_edges_txn(txn: LoggingTransaction) -> bool:
"""Returns True if we're done."""
# first we need to find an endpoint.
txn.execute(
"""
SELECT event_id FROM event_edges
WHERE event_id > ?
ORDER BY event_id
LIMIT 1 OFFSET ?
""",
(last_event_id, batch_size),
)
endpoint = None
row = txn.fetchone()
if row:
endpoint = row[0]
where_clause = "ee.event_id > ?"
args = [last_event_id]
if endpoint:
where_clause += " AND ee.event_id <= ?"
args.append(endpoint)
# now delete any that:
# - have is_state=TRUE, or
# - do not correspond to a row in `events`
txn.execute(
f"""
DELETE FROM event_edges
WHERE event_id IN (
SELECT ee.event_id
FROM event_edges ee
LEFT JOIN events ev USING (event_id)
WHERE ({where_clause}) AND
(is_state OR ev.event_id IS NULL)
)""",
args,
)
logger.info(
"cleaned up event_edges up to %s: removed %i/%i rows",
endpoint,
txn.rowcount,
batch_size,
)
if endpoint is not None:
self.db_pool.updates._background_update_progress_txn(
txn,
_BackgroundUpdates.EVENT_EDGES_DROP_INVALID_ROWS,
{"last_event_id": endpoint},
)
return False
# if that was the final batch, we validate the foreign key.
#
# The constraint should have been in place and enforced for new rows since
# before we started deleting invalid rows, so there's no chance for any
# invalid rows to have snuck in the meantime. In other words, this really
# ought to succeed.
logger.info("cleaned up event_edges; enabling foreign key")
txn.execute(
"ALTER TABLE event_edges VALIDATE CONSTRAINT event_edges_event_id_fkey"
)
return True
done = await self.db_pool.runInteraction(
desc="drop_invalid_event_edges", func=drop_invalid_event_edges_txn
)
if done:
await self.db_pool.updates._end_background_update(
_BackgroundUpdates.EVENT_EDGES_DROP_INVALID_ROWS
)
return batch_size
async def _background_events_populate_state_key_rejections(
self, progress: JsonDict, batch_size: int
) -> int:
synapse/storage/schema/main/delta/95/01_foreground_event_edges_state_cleanup.py
+74
View File
@@ -0,0 +1,74 @@
#
# This file is licensed under the Affero General Public License (AGPL) version 3.
#
# Copyright (C) 2026 Element Creations Ltd
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# See the GNU Affero General Public License for more details:
# <https://www.gnu.org/licenses/agpl-3.0.html>.
#
import logging
from synapse.storage.database import LoggingTransaction
from synapse.storage.engines import BaseDatabaseEngine, Sqlite3Engine
logger = logging.getLogger(__name__)
def run_create(txn: LoggingTransaction, database_engine: BaseDatabaseEngine) -> None:
"""
This migration forces the `event_edges_drop_invalid_rows` background update
scheduled by `71/01rebuild_event_edges.sql.postgres` to be completed
in the foreground if it is still outstanding.
"""
if isinstance(database_engine, Sqlite3Engine):
# SQLite already did everything synchronously in
# `71/01rebuild_event_edges.sql.sqlite`
# So there's nothing to do
return
# Clear the background update whilst also checking it
txn.execute(
"""
DELETE FROM background_updates
WHERE update_name = 'event_edges_drop_invalid_rows'
RETURNING 1
"""
)
if txn.fetchone() is None:
# The background update has completed, so there is nothing to do.
return
logger.warning(
"`event_edges_drop_invalid_rows` has not completed in background; running in foreground!"
)
# now delete any that:
# - have is_state=TRUE, or
# - do not correspond to a row in `events`
txn.execute(
"""
DELETE FROM event_edges
WHERE event_id IN (
SELECT ee.event_id
FROM event_edges ee
LEFT JOIN events ev USING (event_id)
WHERE (is_state OR ev.event_id IS NULL)
)
""",
)
logger.info("Deleted %i legacy state edges from `event_edges` table", txn.rowcount)
logger.info("Enabling foreign key")
txn.execute(
"""
ALTER TABLE event_edges
VALIDATE CONSTRAINT event_edges_event_id_fkey
"""
)
synapse/types/storage/__init__.py
-1
View File
@@ -31,7 +31,6 @@ class _BackgroundUpdates:
INDEX_STREAM_ORDERING2_TS = "index_stream_ordering2_ts"
REPLACE_STREAM_ORDERING_COLUMN = "replace_stream_ordering_column"
EVENT_EDGES_DROP_INVALID_ROWS = "event_edges_drop_invalid_rows"
EVENT_EDGES_REPLACE_INDEX = "event_edges_replace_index"
EVENTS_POPULATE_STATE_KEY_REJECTIONS = "events_populate_state_key_rejections"