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

Force events_populate_state_key_rejections background update to run in foreground

Browse Source
This commit is contained in:
Olivier 'reivilibre
2026-08-10 12:49:26 +01:00
committed by Olivier 'reivilibre
parent 6605853805
commit 97ad4ba810
3 changed files with 64 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/storage/databases/main/events_bg_updates.py
-85
View File
@@ -290,11 +290,6 @@ class EventsBackgroundUpdatesStore(
replaces_index="ev_edges_id",
)
self.db_pool.updates.register_background_update_handler(
_BackgroundUpdates.EVENTS_POPULATE_STATE_KEY_REJECTIONS,
self._background_events_populate_state_key_rejections,
)
# Add an index that would be useful for jumping to date using
# get_event_id_for_timestamp.
self.db_pool.updates.register_background_index_update(
@@ -1471,86 +1466,6 @@ class EventsBackgroundUpdatesStore(
return batch_size
async def _background_events_populate_state_key_rejections(
self, progress: JsonDict, batch_size: int
) -> int:
"""Back-populate `events.state_key` and `events.rejection_reason"""
min_stream_ordering_exclusive = progress["min_stream_ordering_exclusive"]
max_stream_ordering_inclusive = progress["max_stream_ordering_inclusive"]
def _populate_txn(txn: LoggingTransaction) -> bool:
"""Returns True if we're done."""
# first we need to find an endpoint.
# we need to find the final row in the batch of batch_size, which means
# we need to skip over (batch_size-1) rows and get the next row.
txn.execute(
"""
SELECT stream_ordering FROM events
WHERE stream_ordering > ? AND stream_ordering <= ?
ORDER BY stream_ordering
LIMIT 1 OFFSET ?
""",
(
min_stream_ordering_exclusive,
max_stream_ordering_inclusive,
batch_size - 1,
),
)
row = txn.fetchone()
if row:
endpoint = row[0]
else:
# if the query didn't return a row, we must be almost done. We just
# need to go up to the recorded max_stream_ordering.
endpoint = max_stream_ordering_inclusive
where_clause = "stream_ordering > ? AND stream_ordering <= ?"
args = [min_stream_ordering_exclusive, endpoint]
# now do the updates.
txn.execute(
f"""
UPDATE events
SET state_key = (SELECT state_key FROM state_events se WHERE se.event_id = events.event_id),
rejection_reason = (SELECT reason FROM rejections rej WHERE rej.event_id = events.event_id)
WHERE ({where_clause})
""",
args,
)
logger.info(
"populated new `events` columns up to %i/%i: updated %i rows",
endpoint,
max_stream_ordering_inclusive,
txn.rowcount,
)
if endpoint >= max_stream_ordering_inclusive:
# we're done
return True
progress["min_stream_ordering_exclusive"] = endpoint
self.db_pool.updates._background_update_progress_txn(
txn,
_BackgroundUpdates.EVENTS_POPULATE_STATE_KEY_REJECTIONS,
progress,
)
return False
done = await self.db_pool.runInteraction(
desc="events_populate_state_key_rejections", func=_populate_txn
)
if done:
await self.db_pool.updates._end_background_update(
_BackgroundUpdates.EVENTS_POPULATE_STATE_KEY_REJECTIONS
)
return batch_size
async def _sliding_sync_prefill_joined_rooms_to_recalculate_table_bg_update(
self, progress: JsonDict, _batch_size: int
) -> int:
synapse/storage/schema/main/delta/95/03_foreground_events_populate_state_key_rejections.py
+64
View File
@@ -0,0 +1,64 @@
#
# 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 json
import logging
from synapse.storage.database import LoggingTransaction
from synapse.storage.engines import BaseDatabaseEngine
logger = logging.getLogger(__name__)
def run_create(txn: LoggingTransaction, database_engine: BaseDatabaseEngine) -> None:
"""
This migration forces the `events_populate_state_key_rejections` background
update scheduled by `72/03bg_populate_events_columns.py` to be completed
in the foreground if it is still outstanding.
"""
# Clear the background update whilst also checking it
txn.execute(
"""
DELETE FROM background_updates
WHERE update_name = 'events_populate_state_key_rejections'
RETURNING progress_json
"""
)
row = txn.fetchone()
if row is None:
# The background update has completed, so there is nothing to do.
return
progress = json.loads(row[0])
min_stream_ordering_exclusive = progress["min_stream_ordering_exclusive"]
max_stream_ordering_inclusive = progress["max_stream_ordering_inclusive"]
logger.warning(
"`events_populate_state_key_rejections` has not completed in background; running in foreground!"
)
# Backpopulate the `state_key` and `rejection_reason` columns according to the unpopulated range
# specified in the background update progress dict.
# For simplicity, do this in one query.
txn.execute(
"""
UPDATE events
SET state_key = (SELECT state_key FROM state_events se WHERE se.event_id = events.event_id),
rejection_reason = (SELECT reason FROM rejections rej WHERE rej.event_id = events.event_id)
WHERE ? < stream_ordering AND stream_ordering <= ?
""",
(min_stream_ordering_exclusive, max_stream_ordering_inclusive),
)
logger.info("Populated new `events` columns for %i rows", txn.rowcount)
synapse/types/storage/__init__.py
-2
View File
@@ -34,8 +34,6 @@ class _BackgroundUpdates:
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"
EVENTS_JUMP_TO_DATE_INDEX = "events_jump_to_date_index"
CURRENT_STATE_EVENTS_STREAM_ORDERING_INDEX_UPDATE_NAME = (