Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/20069.misc
Original file line number Diff line number Diff line change
@@ -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.
12 changes: 1 addition & 11 deletions synapse/storage/databases/main/event_federation.py
Original file line number Diff line number Diff line change
Expand Up @@ -1503,15 +1503,6 @@ def get_backfill_points_in_room_txn(
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
Expand Down Expand Up @@ -1890,7 +1881,7 @@ def _get_connected_prev_event_backfill_results_txn(
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 */
Expand Down Expand Up @@ -2237,7 +2228,6 @@ def _get_missing_events(
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)),
Expand Down
101 changes: 0 additions & 101 deletions synapse/storage/databases/main/events_bg_updates.py
Original file line number Diff line number Diff line change
Expand Up @@ -280,11 +280,6 @@ def __init__(

################################################################################

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",
Expand Down Expand Up @@ -1497,102 +1492,6 @@ def process(txn: Cursor) -> None:

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:
Expand Down
1 change: 1 addition & 0 deletions synapse/storage/schema/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,7 @@
- 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).
"""


Expand Down
Original file line number Diff line number Diff line change
@@ -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
"""
)
1 change: 0 additions & 1 deletion synapse/types/storage/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
Loading