Fix retrying failed event persistence (#20148)

If we failed to persist an event and then retried, this would fail due
to a conflict error trying to reinsert the state groups that we are
persisting.

This blocks all new events being persisted in the room until the
instance is restarted (which clears entries from the table matching the
instance name). This is because we always mark the [latest persisted
state group in the room as being
persisted](https://github.com/element-hq/synapse/blob/781d08df263e640d85ca5c8b5ff514433e019c6f/synapse/storage/databases/state/deletion.py#L242-L243).

Also add an `inserted_ts` field which we use to detect stale entries,
e.g. from instances that have been taken offline permanently. These
aren't a problem beyond blocking deletion of the associated state groups

Introduced in https://github.com/element-hq/synapse/pull/18107

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: Andrew Morgan <1342360+anoadragon453@users.noreply.github.com>

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.

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(

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.
 """
 
 

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:
+-- <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+-- 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;

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:
+-- <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+-- 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;

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."""