mirror of
https://github.com/element-hq/synapse.git
synced 2026-10-23 19:33:44 +00:00
Stop reading from event_edges.is_state to prepare for dropping it. (#20069)
Gets one step further in the job started by https://github.com/matrix-org/synapse/pull/12893 Part of: https://github.com/element-hq/synapse/issues/20026 ## Manual testing strategy Unfortunately it's quite difficult to actually trigger this. You need to: - check out an old full_schema, e.g. 54 which was present when `v1.62.0` was released - remove the version 72 `full_schema`, as it already has the background update applied - remove some deltas that are broken when upgrading across such a large version gap (!) https://github.com/element-hq/synapse/issues/20070 - then run any trial test with Postgres enabled to prove that the database schema applies ``` git checkout v1.62.0 -- synapse/storage/schema/state/full_schemas/54 git checkout v1.62.0 -- synapse/storage/schema/main/full_schemas/54 git checkout v1.62.0 -- synapse/storage/schema/common/full_schemas/54 rm -r synapse/storage/schema/*/full_schemas/72 rm synapse/storage/schema/main/delta/58/10_pushrules_enabled_delete_obsolete.sql rm synapse/storage/schema/main/delta/79/04_mitigate_stream_ordering_update_race.py rm synapse/storage/schema/main/delta/56/redaction_censor3_fix_update.sql.postgres rm synapse/storage/schema/main/delta/60/01recreate_stream_ordering.sql.postgres SYNAPSE_TEST_LOG_LEVEL=DEBUG SYNAPSE_POSTGRES=1 poetry run trial tests.handlers.test_sliding_sync.ComputeInterestedRoomsTestCase_new.test_no_joins_after_to_token ``` As light validation, I purposefully inserted a SQL error into the query to check that it is caught by the above testing strategy. --------- Signed-off-by: Olivier 'reivilibre <oliverw@matrix.org>
This commit is contained in:
1 parent
766f057305
commit
636c6a3812
6 files changed
+77
-113
No files matched your search
@@ -0,0 +1 @@
|
||||
Stop reading from `event_edges.is_state` to prepare for dropping it. The column has been obsolete since Synapse 1.62.0.
|
||||
@@ -1503,15 +1503,6 @@ class EventFederationWorkerStore(
|
||||
AND failed_backfill_attempt_info.event_id = backward_extrem.event_id
|
||||
WHERE
|
||||
backward_extrem.room_id = ?
|
||||
/* We only care about non-state edges because we used to use
|
||||
* `event_edges` for two different sorts of "edges" (the current
|
||||
* event DAG, but also a link to the previous state, for state
|
||||
* events). These legacy state event edges can be distinguished by
|
||||
* `is_state` and are removed from the codebase and schema but
|
||||
* because the schema change is in a background update, it's not
|
||||
* necessarily safe to assume that it will have been completed.
|
||||
*/
|
||||
AND edge.is_state is FALSE
|
||||
/**
|
||||
* We only want backwards extremities that are older than or at
|
||||
* the same position of the given `nearby_depth` (where older
|
||||
@@ -1890,7 +1881,7 @@ class EventFederationWorkerStore(
|
||||
WHERE NOT events.outlier
|
||||
|
||||
/* Look for an edge which matches the given event_id */
|
||||
AND event_edges.event_id = ? AND NOT event_edges.is_state
|
||||
AND event_edges.event_id = ?
|
||||
|
||||
/* Because we can have many events at the same depth,
|
||||
* we want to also tie-break and sort on stream_ordering */
|
||||
@@ -2237,7 +2228,6 @@ class EventFederationWorkerStore(
|
||||
JOIN events ON events.event_id = ee.prev_event_id
|
||||
WHERE ee.event_id = ?
|
||||
AND events.room_id = ?
|
||||
AND NOT ee.is_state
|
||||
LIMIT ?
|
||||
""",
|
||||
(event_id, room_id, limit - len(event_results)),
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -182,6 +182,7 @@ Changes in SCHEMA_VERSION = 95
|
||||
- MSC4140: Stop using the `delayed_events_stream_pos` table, now that
|
||||
delayed state events are no longer canceled by more recent state events
|
||||
sent from a different user.
|
||||
- `event_edges.is_state` is no longer read from (it has not been written to since SCHEMA_VERSION 72).
|
||||
"""
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
"""
|
||||
)
|
||||
@@ -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"
|
||||
|
||||
Reference in new issue
Block a user