diff --git a/changelog.d/20148.bugfix b/changelog.d/20148.bugfix new file mode 100644 index 0000000000..2983484208 --- /dev/null +++ b/changelog.d/20148.bugfix @@ -0,0 +1 @@ +Fix a bug where, until Synapse was restarted, new events in a room would fail to be persisted if the database went down while an event was being persisted in the room. Bug introduced in v1.124.0. diff --git a/synapse/storage/databases/state/deletion.py b/synapse/storage/databases/state/deletion.py index 23150e8626..9f08f570d8 100644 --- a/synapse/storage/databases/state/deletion.py +++ b/synapse/storage/databases/state/deletion.py @@ -14,6 +14,7 @@ import contextlib +import logging from typing import ( TYPE_CHECKING, AbstractSet, @@ -30,11 +31,14 @@ from synapse.storage.database import ( make_in_list_sql_clause, ) from synapse.storage.engines import PostgresEngine +from synapse.util.duration import Duration from synapse.util.stringutils import shortstr if TYPE_CHECKING: from synapse.server import HomeServer +logger = logging.getLogger(__name__) + class StateDeletionDataStore: """Manages deletion of state groups in a safe manner. @@ -81,6 +85,12 @@ class StateDeletionDataStore: # event will fail to persist (as well as any event in the same batch). DELAY_BEFORE_DELETION_MS = 10 * 60 * 1000 + # How old a row in `state_groups_persisting` has to be before we assume the + # persist that wrote it has gone away. This should be much longer than any + # persist can take. If we clear the row of a live persist, then its state + # groups can be deleted while it is still using them. + STALE_PERSISTING_DURATION = Duration(days=7) + def __init__( self, database: DatabasePool, @@ -91,17 +101,35 @@ class StateDeletionDataStore: self.db_pool = database self._instance_name = hs.get_instance_name() - with db_conn.cursor(txn_name="_clear_existing_persising") as txn: - self._clear_existing_persising(txn) + with db_conn.cursor(txn_name="_clear_existing_persisting") as txn: + self._clear_existing_persisting(txn) - def _clear_existing_persising(self, txn: LoggingTransaction) -> None: + def _clear_existing_persisting(self, txn: LoggingTransaction) -> None: """On startup we clear any entries in `state_groups_persisting` that - match our instance name, in case of a previous unclean shutdown""" + match our instance name (or are very old), in case of a previous unclean + shutdown.""" - self.db_pool.simple_delete_txn( - txn, - table="state_groups_persisting", - keyvalues={"instance_name": self._instance_name}, + # Delete any rows that are very old, or that match our instance name. We + # clear all stale rows, even if they don't match our instance name, as + # we don't know if the instance that created them is still running. + cutoff = self._clock.time_msec() - self.STALE_PERSISTING_DURATION.as_millis() + sql = """ + DELETE FROM state_groups_persisting + WHERE inserted_ts < ? OR (instance_name = ?) + RETURNING state_group + """ + txn.execute(sql, (cutoff, self._instance_name)) + + # Two instances can each have a stale row for the same state group. + state_groups = {state_group for (state_group,) in txn} + + if not state_groups: + return + + logger.info( + "Cleared %d stale state groups from state_groups_persisting: %s", + len(state_groups), + shortstr(state_groups), ) async def check_state_groups_and_bump_deletion( @@ -277,11 +305,20 @@ class StateDeletionDataStore: f"state groups have been deleted: {shortstr(missing_state_groups)}" ) - self.db_pool.simple_insert_many_txn( + # There is a unique key on (state_group, instance_name) so we need to + # handle the case where we have already marked a state group as being + # persisted. This can happen if we fail to persist an event and then + # retry. + now = self._clock.time_msec() + self.db_pool.simple_upsert_many_txn( txn, table="state_groups_persisting", - keys=("state_group", "instance_name"), - values=[(state_group, self._instance_name) for state_group in state_groups], + key_names=("state_group", "instance_name"), + key_values=[ + (state_group, self._instance_name) for state_group in state_groups + ], + value_names=("inserted_ts",), + value_values=[(now,) for _ in state_groups], ) def _finish_persisting_txn( diff --git a/synapse/storage/schema/__init__.py b/synapse/storage/schema/__init__.py index 0d9022116d..926121d8e1 100644 --- a/synapse/storage/schema/__init__.py +++ b/synapse/storage/schema/__init__.py @@ -176,6 +176,7 @@ Changes in SCHEMA_VERSION = 94 - Add `recheck` column (boolean, default true) to the `redactions` table. - MSC4242: Add state DAG tables. - MSC4429/MSC4262: Track updates to user profile fields via a new stream. + - Add an `inserted_ts` column to the `state_groups_persisting` table. """ diff --git a/synapse/storage/schema/state/delta/94/01_state_groups_persisting_inserted_ts.sql.postgres b/synapse/storage/schema/state/delta/94/01_state_groups_persisting_inserted_ts.sql.postgres new file mode 100644 index 0000000000..4e5c0eea96 --- /dev/null +++ b/synapse/storage/schema/state/delta/94/01_state_groups_persisting_inserted_ts.sql.postgres @@ -0,0 +1,22 @@ +-- +-- 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: +-- . + +-- Record when we marked a state group as being persisted, so that we can clear +-- out rows left behind by a persist that never finished. +-- +-- The rows already in the table are stamped with the time this runs, so they get +-- the same grace period as a new row before we treat them as stale. The default +-- covers instances that haven't been upgraded yet, which insert without the +-- column, and is what lets the column be NOT NULL during a rolling upgrade. +ALTER TABLE state_groups_persisting + ADD COLUMN inserted_ts BIGINT NOT NULL DEFAULT extract(epoch from now()) * 1000; diff --git a/synapse/storage/schema/state/delta/94/01_state_groups_persisting_inserted_ts.sql.sqlite b/synapse/storage/schema/state/delta/94/01_state_groups_persisting_inserted_ts.sql.sqlite new file mode 100644 index 0000000000..7720a814a5 --- /dev/null +++ b/synapse/storage/schema/state/delta/94/01_state_groups_persisting_inserted_ts.sql.sqlite @@ -0,0 +1,31 @@ +-- +-- 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: +-- . + +-- Record when we marked a state group as being persisted, so that we can clear +-- out rows left behind by a persist that never finished. +-- +-- On SQLite we must be in monolith mode, so every row already in the table was +-- written by this process before it restarted, and is dead. Zero marks them as +-- stale straight away, which is what we want. Contrast Postgres, where such a +-- row may belong to a worker that is still running and so gets a grace period. +-- +-- Note that this also correctly handles the case of Synapse version rolling +-- back, as new rows will be written with a 0 inserted_ts, *but* since its +-- running with the old code nothing deletes those row. On upgrade the restart +-- will mean any rows are automatically stale as above. +-- +-- The default only applies to an insert that omits the column, which our own +-- code never does. SQLite only accepts a constant default in ADD COLUMN, so it +-- could not be the current time in any case. +ALTER TABLE state_groups_persisting + ADD COLUMN inserted_ts BIGINT NOT NULL DEFAULT 0; diff --git a/tests/storage/test_state_deletion.py b/tests/storage/test_state_deletion.py index d4079c372e..066d69c93d 100644 --- a/tests/storage/test_state_deletion.py +++ b/tests/storage/test_state_deletion.py @@ -14,12 +14,15 @@ import logging +from collections.abc import Collection +from unittest.mock import patch from twisted.internet.testing import MemoryReactor from synapse.rest import admin from synapse.rest.client import login, room from synapse.server import HomeServer +from synapse.storage.database import LoggingTransaction from synapse.util.clock import Clock from tests.test_utils.event_injection import create_event @@ -48,8 +51,20 @@ class StateDeletionStoreTestCase(HomeserverTestCase): self.purge_events._delete_state_loop_call.stop() self.user_id = self.register_user("test", "password") - tok = self.login("test", "password") - self.room_id = self.helper.create_room_as(self.user_id, tok=tok) + self.tok = self.login("test", "password") + self.room_id = self.helper.create_room_as(self.user_id, tok=self.tok) + + def get_persisting_marker_rows(self) -> list[tuple[int, str, int]]: + """Return the contents of the `state_groups_persisting` table.""" + + return self.get_success( + self.state_deletion_store.db_pool.simple_select_list( + table="state_groups_persisting", + keyvalues=None, + retcols=("state_group", "instance_name", "inserted_ts"), + desc="get_persisting_marker_rows", + ) + ) def check_if_can_be_deleted(self, state_group: int) -> bool: """Check if the state group is pending deletion.""" @@ -107,6 +122,39 @@ class StateDeletionStoreTestCase(HomeserverTestCase): self.get_success(ctx_mgr.__aenter__()) self.get_success(ctx_mgr.__aexit__(Exception, Exception("test"), None)) + def test_retry_send_after_failed_clean_up(self) -> None: + """Test that we can retry sending an event after the clean up of the + `state_groups_persisting` rows failed, as it does when the database goes + away while we're persisting.""" + + fail_clean_up = True + orig_finish_persisting_txn = self.state_deletion_store._finish_persisting_txn + + def _finish_persisting_txn( + txn: LoggingTransaction, state_groups: Collection[int], error: bool + ) -> None: + if fail_clean_up: + raise Exception("Database has gone away") + + orig_finish_persisting_txn(txn, state_groups, error) + + with patch.object( + self.state_deletion_store, + "_finish_persisting_txn", + new=_finish_persisting_txn, + ): + self.helper.send(self.room_id, body="first", tok=self.tok, expect_code=500) + + # The rows are still in place, as the transaction that would have + # removed them was rolled back. + self.assertNotEqual(self.get_persisting_marker_rows(), []) + + # The database has come back, so the retry should now go through. + fail_clean_up = False + self.helper.send(self.room_id, body="second", tok=self.tok) + + self.assertEqual(self.get_persisting_marker_rows(), []) + def test_existing_pending_deletion_is_cleared(self) -> None: """Test that the pending deletion flag gets cleared when the state group gets persisted."""