From f07b9de38cd9e1c9859eaa0aba941ae1196a0bdb Mon Sep 17 00:00:00 2001 From: SpiritCroc Date: Fri, 21 Aug 2026 13:59:51 +0200 Subject: [PATCH 1/3] PLAT-38547 Account-data compare-and-swap --- synapse/api/errors.py | 3 + synapse/handlers/account_data.py | 16 ++- synapse/replication/http/account_data.py | 20 +++- synapse/rest/client/account_data.py | 14 ++- .../storage/databases/main/account_data.py | 110 +++++++++++++++++- 5 files changed, 144 insertions(+), 19 deletions(-) diff --git a/synapse/api/errors.py b/synapse/api/errors.py index 3dfea09b8..c1a7baeb9 100644 --- a/synapse/api/errors.py +++ b/synapse/api/errors.py @@ -152,6 +152,9 @@ class Codes(str, Enum): # Part of MSC4326 UNKNOWN_DEVICE = "ORG.MATRIX.MSC4326.M_UNKNOWN_DEVICE" + # Beeper: account data compare-and-swap via com.beeper.expect_revision_id + EXPECTED_REVISION_ID_MISMATCH = "COM.BEEPER.REVISION_ID_MISMATCH" + class CodeMessageException(RuntimeError): """An exception with integer code, a message string attributes and optional headers. diff --git a/synapse/handlers/account_data.py b/synapse/handlers/account_data.py index c6168377e..1002a4394 100644 --- a/synapse/handlers/account_data.py +++ b/synapse/handlers/account_data.py @@ -104,7 +104,8 @@ async def _notify_modules( logger.exception("Failed to run module callback %s: %s", callback, e) async def add_account_data_to_room( - self, user_id: str, room_id: str, account_data_type: str, content: JsonDict + self, user_id: str, room_id: str, account_data_type: str, content: JsonDict, + expected_revision_id: str | None = None, ) -> int: """Add some account_data to a room for a user. @@ -113,13 +114,15 @@ async def add_account_data_to_room( room_id: The room to add a tag for. account_data_type: The type of account_data to add. content: A json object to associate with the tag. + expected_revision_id: If set, only write if the stored content's + `com.beeper.revision_id` matches (compare-and-swap). Returns: The maximum stream ID. """ if self._instance_name in self._account_data_writers: max_stream_id = await self._store.add_account_data_to_room( - user_id, room_id, account_data_type, content + user_id, room_id, account_data_type, content, expected_revision_id ) self._notifier.on_new_event( @@ -136,6 +139,7 @@ async def add_account_data_to_room( room_id=room_id, account_data_type=account_data_type, content=content, + expected_revision_id=expected_revision_id, ) return response["max_stream_id"] @@ -181,7 +185,8 @@ async def remove_account_data_for_room( return response["max_stream_id"] async def add_account_data_for_user( - self, user_id: str, account_data_type: str, content: JsonDict + self, user_id: str, account_data_type: str, content: JsonDict, + expected_revision_id: str | None = None, ) -> int: """Add some global account_data for a user. @@ -189,6 +194,8 @@ async def add_account_data_for_user( user_id: The user to add some account data for. account_data_type: The type of account_data to add. content: The content json dictionary. + expected_revision_id: If set, only write if the stored content's + `com.beeper.revision_id` matches (compare-and-swap). Returns: The maximum stream ID. @@ -196,7 +203,7 @@ async def add_account_data_for_user( if self._instance_name in self._account_data_writers: max_stream_id = await self._store.add_account_data_for_user( - user_id, account_data_type, content + user_id, account_data_type, content, expected_revision_id ) self._notifier.on_new_event( @@ -212,6 +219,7 @@ async def add_account_data_for_user( user_id=user_id, account_data_type=account_data_type, content=content, + expected_revision_id=expected_revision_id, ) return response["max_stream_id"] diff --git a/synapse/replication/http/account_data.py b/synapse/replication/http/account_data.py index 560973b91..770af4fc3 100644 --- a/synapse/replication/http/account_data.py +++ b/synapse/replication/http/account_data.py @@ -58,11 +58,14 @@ def __init__(self, hs: "HomeServer"): @staticmethod async def _serialize_payload( # type: ignore[override] - user_id: str, account_data_type: str, content: JsonDict + user_id: str, account_data_type: str, content: JsonDict, + expected_revision_id: str | None = None, ) -> JsonDict: - payload = { + payload: JsonDict = { "content": content, } + if expected_revision_id is not None: + payload["expected_revision_id"] = expected_revision_id return payload @@ -70,7 +73,8 @@ async def _handle_request( # type: ignore[override] self, request: Request, content: JsonDict, user_id: str, account_data_type: str ) -> tuple[int, JsonDict]: max_stream_id = await self.handler.add_account_data_for_user( - user_id, account_data_type, content["content"] + user_id, account_data_type, content["content"], + expected_revision_id=content.get("expected_revision_id"), ) return 200, {"max_stream_id": max_stream_id} @@ -138,11 +142,14 @@ def __init__(self, hs: "HomeServer"): @staticmethod async def _serialize_payload( # type: ignore[override] - user_id: str, room_id: str, account_data_type: str, content: JsonDict + user_id: str, room_id: str, account_data_type: str, content: JsonDict, + expected_revision_id: str | None = None, ) -> JsonDict: - payload = { + payload: JsonDict = { "content": content, } + if expected_revision_id is not None: + payload["expected_revision_id"] = expected_revision_id return payload @@ -155,7 +162,8 @@ async def _handle_request( # type: ignore[override] account_data_type: str, ) -> tuple[int, JsonDict]: max_stream_id = await self.handler.add_account_data_to_room( - user_id, room_id, account_data_type, content["content"] + user_id, room_id, account_data_type, content["content"], + expected_revision_id=content.get("expected_revision_id"), ) return 200, {"max_stream_id": max_stream_id} diff --git a/synapse/rest/client/account_data.py b/synapse/rest/client/account_data.py index 5bcdd907b..fbefa3e62 100644 --- a/synapse/rest/client/account_data.py +++ b/synapse/rest/client/account_data.py @@ -26,7 +26,11 @@ from synapse.api.constants import AccountDataTypes, ReceiptTypes from synapse.api.errors import AuthError, Codes, NotFoundError, SynapseError from synapse.http.server import HttpServer -from synapse.http.servlet import RestServlet, parse_json_object_from_request +from synapse.http.servlet import ( + RestServlet, + parse_json_object_from_request, + parse_string, +) from synapse.http.site import SynapseRequest from synapse.rest.client.read_marker import ReadMarkerRestServlet from synapse.types import JsonDict, JsonMapping, RoomID @@ -86,6 +90,7 @@ async def on_PUT( _check_can_set_account_data_type(account_data_type) body = parse_json_object_from_request(request) + expected_revision_id = parse_string(request, "com.beeper.expect_revision_id") # If experimental support for MSC3391 is enabled, then providing an empty dict # as the value for an account data type should be functionally equivalent to @@ -97,7 +102,9 @@ async def on_PUT( ) return 200, {} - await self.handler.add_account_data_for_user(user_id, account_data_type, body) + await self.handler.add_account_data_for_user( + user_id, account_data_type, body, expected_revision_id + ) return 200, {} @@ -209,6 +216,7 @@ async def on_PUT( _check_can_set_account_data_type(account_data_type) body = parse_json_object_from_request(request) + expected_revision_id = parse_string(request, "com.beeper.expect_revision_id") # If experimental support for MSC3391 is enabled, then providing an empty dict # as the value for an account data type should be functionally equivalent to @@ -221,7 +229,7 @@ async def on_PUT( return 200, {} await self.handler.add_account_data_to_room( - user_id, room_id, account_data_type, body + user_id, room_id, account_data_type, body, expected_revision_id ) return 200, {} diff --git a/synapse/storage/databases/main/account_data.py b/synapse/storage/databases/main/account_data.py index f4706487a..ada70541f 100644 --- a/synapse/storage/databases/main/account_data.py +++ b/synapse/storage/databases/main/account_data.py @@ -41,6 +41,7 @@ ) from synapse.storage.databases.main.cache import CacheInvalidationWorkerStore from synapse.storage.databases.main.push_rule import PushRulesWorkerStore +from synapse.storage.engines import PostgresEngine from synapse.storage.invite_rule import ( AllowAllInviteRulesConfig, InviteRulesConfig, @@ -640,7 +641,8 @@ def process_replication_position( super().process_replication_position(stream_name, instance_name, token) async def add_account_data_to_room( - self, user_id: str, room_id: str, account_data_type: str, content: JsonDict + self, user_id: str, room_id: str, account_data_type: str, content: JsonDict, + expected_revision_id: str | None = None, ) -> int: """Add some account_data to a room for a user. @@ -649,6 +651,8 @@ async def add_account_data_to_room( room_id: The room to add a tag for. account_data_type: The type of account_data to add. content: A json object to associate with the tag. + expected_revision_id: If set, only write if the stored content's + `com.beeper.revision_id` matches (compare-and-swap). Returns: The maximum stream ID. @@ -657,9 +661,11 @@ async def add_account_data_to_room( content_json = json_encoder.encode(content) - async with self._account_data_id_gen.get_next() as next_id: - await self.db_pool.simple_upsert( - desc="add_room_account_data", + def _add_account_data_to_room_txn( + txn: LoggingTransaction, next_id: int + ) -> None: + self._upsert_account_data_txn( + txn, table="room_account_data", keyvalues={ "user_id": user_id, @@ -667,6 +673,14 @@ async def add_account_data_to_room( "account_data_type": account_data_type, }, values={"stream_id": next_id, "content": content_json}, + expected_revision_id=expected_revision_id, + ) + + async with self._account_data_id_gen.get_next() as next_id: + await self.db_pool.runInteraction( + "add_room_account_data", + _add_account_data_to_room_txn, + next_id, ) self._account_data_stream_cache.entity_has_changed(user_id, next_id) @@ -742,7 +756,8 @@ def _remove_account_data_for_room_txn( return self._account_data_id_gen.get_current_token() async def add_account_data_for_user( - self, user_id: str, account_data_type: str, content: JsonDict + self, user_id: str, account_data_type: str, content: JsonDict, + expected_revision_id: str | None = None, ) -> int: """Add some global account_data for a user. @@ -750,6 +765,8 @@ async def add_account_data_for_user( user_id: The user to add a tag for. account_data_type: The type of account_data to add. content: A json object to associate with the tag. + expected_revision_id: If set, only write if the stored content's + `com.beeper.revision_id` matches (compare-and-swap). Returns: The maximum stream ID. @@ -764,6 +781,7 @@ async def add_account_data_for_user( user_id, account_data_type, content, + expected_revision_id, ) self._account_data_stream_cache.entity_has_changed(user_id, next_id) @@ -774,6 +792,84 @@ async def add_account_data_for_user( return self._account_data_id_gen.get_current_token() + def _upsert_account_data_txn( + self, + txn: LoggingTransaction, + table: str, + keyvalues: dict[str, str], + values: dict[str, Any], + expected_revision_id: str | None, + ) -> None: + """Beeper: upsert account data, enforcing the compare-and-swap + condition when an expected revision ID is given. + + Raises a 409 SynapseError if the stored content has a (string) + `com.beeper.revision_id` that differs from the expected one. A missing + row, missing field, or non-string field matches any expected value. + """ + if expected_revision_id is None: + self.db_pool.simple_upsert_txn(txn, table, keyvalues, values) + return + + select_sql = "SELECT content FROM %s WHERE %s" % ( + table, + " AND ".join("%s = ?" % k for k in keyvalues), + ) + if isinstance(self.database_engine, PostgresEngine): + # Lock the row so concurrent CAS writes serialize. Under Synapse's + # default REPEATABLE READ isolation, a row modified by a concurrent + # transaction instead raises a serialization failure, which + # runInteraction retries with a fresh snapshot. (SQLite serializes + # writes anyway, so a plain SELECT suffices there.) + select_sql += " FOR UPDATE" + txn.execute(select_sql, list(keyvalues.values())) + row = txn.fetchone() + + if row is None: + # There is no row to lock, so a check-then-write would let two + # concurrent first writes both pass the check. Make the INSERT + # itself the atomic point instead: it only succeeds if no + # concurrent write landed first (in-flight inserts on the same + # key serialize via speculative insertion), so on success the + # no-data match genuinely held at write time. On conflict, fall + # through to compare against the winning row. + if self.db_pool.simple_upsert_txn_native_upsert( + txn, table, keyvalues, values={}, insertion_values=values + ): + return + txn.execute(select_sql, list(keyvalues.values())) + row = txn.fetchone() + if row is None: + # Under REPEATABLE READ the conflicting row was committed + # after our snapshot, so the re-read cannot see it. The + # upsert below then hits that invisible row and raises a + # serialization failure (40001), which runInteraction + # retries from scratch with a fresh snapshot that does see + # the row and compares against it. (Under READ COMMITTED + # the re-read would have found the row directly.) + self.db_pool.simple_upsert_txn(txn, table, keyvalues, values) + return + + stored_content = db_to_json(row[0]) + stored_revision_id = None + if isinstance(stored_content, dict): + rev = stored_content.get("com.beeper.revision_id") + if isinstance(rev, str): + stored_revision_id = rev + + if ( + stored_revision_id is not None + and stored_revision_id != expected_revision_id + ): + raise SynapseError( + 409, + "Account data revision ID mismatch", + Codes.EXPECTED_REVISION_ID_MISMATCH, + additional_fields={"com.beeper.current_content": stored_content}, + ) + + self.db_pool.simple_upsert_txn(txn, table, keyvalues, values) + def _add_account_data_for_user( self, txn: LoggingTransaction, @@ -781,6 +877,7 @@ def _add_account_data_for_user( user_id: str, account_data_type: str, content: JsonDict, + expected_revision_id: str | None = None, ) -> None: content_json = json_encoder.encode(content) @@ -796,11 +893,12 @@ def _add_account_data_for_user( ) } - self.db_pool.simple_upsert_txn( + self._upsert_account_data_txn( txn, table="account_data", keyvalues={"user_id": user_id, "account_data_type": account_data_type}, values={"stream_id": next_id, "content": content_json}, + expected_revision_id=expected_revision_id, ) # Ignored users get denormalized into a separate table as an optimisation. From ee7c381ea8a3dbb61e8901b16b40cfbc0a742555 Mon Sep 17 00:00:00 2001 From: SpiritCroc Date: Fri, 21 Aug 2026 14:00:16 +0200 Subject: [PATCH 2/3] Claude's tests --- tests/rest/client/test_account_data.py | 173 +++++++++++++++++++++++++ tests/storage/test_account_data.py | 143 +++++++++++++++++++- 2 files changed, 315 insertions(+), 1 deletion(-) diff --git a/tests/rest/client/test_account_data.py b/tests/rest/client/test_account_data.py index aff873d87..73d19115f 100644 --- a/tests/rest/client/test_account_data.py +++ b/tests/rest/client/test_account_data.py @@ -18,13 +18,21 @@ # [This file includes modifications made by New Vector Limited] # # +import urllib.parse from unittest.mock import AsyncMock +from twisted.internet.testing import MemoryReactor + from synapse.api.constants import ReceiptTypes +from synapse.api.errors import Codes from synapse.rest import admin from synapse.rest.client import account_data, login, room +from synapse.server import HomeServer +from synapse.types import JsonDict +from synapse.util.clock import Clock from tests import unittest +from tests.server import FakeChannel class AccountDataTestCase(unittest.HomeserverTestCase): @@ -212,3 +220,168 @@ def test_beeper_inbox_state_endpoint_can_set_read_marker(self) -> None: ) ) self.assertNotEqual(existing_read_marker, new_read_marker) + + +class AccountDataCASTestCase(unittest.HomeserverTestCase): + """Tests for the com.beeper.expect_revision_id compare-and-swap query param.""" + + servlets = [ + admin.register_servlets, + login.register_servlets, + room.register_servlets, + account_data.register_servlets, + ] + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.store = hs.get_datastores().main + self.user_id = self.register_user("user", "password") + self.tok = self.login("user", "password") + self.room_id = self.helper.create_room_as(self.user_id, tok=self.tok) + + # (name, PUT path) for both flavors of account data. + self.endpoints = [ + ("global", f"/user/{self.user_id}/account_data"), + ("room", f"/user/{self.user_id}/rooms/{self.room_id}/account_data"), + ] + + def _put( + self, + base_path: str, + account_data_type: str, + content: JsonDict, + expect_revision_id: str | None = None, + ) -> FakeChannel: + url = f"{base_path}/{account_data_type}" + if expect_revision_id is not None: + url += "?com.beeper.expect_revision_id=" + urllib.parse.quote( + expect_revision_id + ) + return self.make_request("PUT", url, content, access_token=self.tok) + + def _get_stored(self, name: str, account_data_type: str) -> JsonDict | None: + if name == "global": + content = self.get_success( + self.store.get_global_account_data_by_type_for_user( + self.user_id, account_data_type + ) + ) + else: + content = self.get_success( + self.store.get_account_data_for_room_and_type( + self.user_id, self.room_id, account_data_type + ) + ) + return dict(content) if content is not None else None + + def test_no_param_always_writes(self) -> None: + """Without the query param, writes succeed regardless of stored revision.""" + for name, path in self.endpoints: + with self.subTest(endpoint=name): + channel = self._put( + path, "org.example.foo", {"com.beeper.revision_id": "abc"} + ) + self.assertEqual(channel.code, 200, channel.result) + + channel = self._put(path, "org.example.foo", {"bar": "baz"}) + self.assertEqual(channel.code, 200, channel.result) + self.assertEqual( + self._get_stored(name, "org.example.foo"), {"bar": "baz"} + ) + + def test_expect_with_no_existing_data(self) -> None: + """Any expected revision (including empty) matches when no data exists.""" + for name, path in self.endpoints: + with self.subTest(endpoint=name): + channel = self._put( + path, "org.example.new1", {"a": 1}, expect_revision_id="anything" + ) + self.assertEqual(channel.code, 200, channel.result) + + channel = self._put( + path, "org.example.new2", {"a": 1}, expect_revision_id="" + ) + self.assertEqual(channel.code, 200, channel.result) + + def test_expect_with_no_stored_revision(self) -> None: + """Any expected revision matches when stored content lacks a revision id.""" + for name, path in self.endpoints: + with self.subTest(endpoint=name): + self._put(path, "org.example.foo", {"bar": "baz"}) + + channel = self._put( + path, "org.example.foo", {"a": 1}, expect_revision_id="xyz" + ) + self.assertEqual(channel.code, 200, channel.result) + + def test_expect_with_non_string_stored_revision(self) -> None: + """A non-string stored revision id is treated as unset.""" + for name, path in self.endpoints: + with self.subTest(endpoint=name): + self._put(path, "org.example.foo", {"com.beeper.revision_id": 5}) + + channel = self._put( + path, "org.example.foo", {"a": 1}, expect_revision_id="xyz" + ) + self.assertEqual(channel.code, 200, channel.result) + + def test_expect_match(self) -> None: + """A matching expected revision allows the write; new content need not + carry a revision id itself.""" + for name, path in self.endpoints: + with self.subTest(endpoint=name): + self._put( + path, + "org.example.foo", + {"com.beeper.revision_id": "abc", "v": 1}, + ) + + new_content = {"com.beeper.revision_id": "def", "v": 2} + channel = self._put( + path, "org.example.foo", new_content, expect_revision_id="abc" + ) + self.assertEqual(channel.code, 200, channel.result) + self.assertEqual(self._get_stored(name, "org.example.foo"), new_content) + + # A revision id in the new content is not required. + channel = self._put( + path, "org.example.foo", {"v": 3}, expect_revision_id="def" + ) + self.assertEqual(channel.code, 200, channel.result) + self.assertEqual(self._get_stored(name, "org.example.foo"), {"v": 3}) + + def test_expect_mismatch(self) -> None: + """A mismatched expected revision is rejected with 409 and the stored + content is returned in the error body.""" + stored_content = {"com.beeper.revision_id": "abc", "v": 1} + for name, path in self.endpoints: + with self.subTest(endpoint=name): + self._put(path, "org.example.foo", stored_content) + + channel = self._put( + path, "org.example.foo", {"v": 2}, expect_revision_id="xyz" + ) + self.assertEqual(channel.code, 409, channel.result) + self.assertEqual( + channel.json_body["errcode"], Codes.EXPECTED_REVISION_ID_MISMATCH + ) + self.assertEqual( + channel.json_body["com.beeper.current_content"], stored_content + ) + # The stored data is unchanged. + self.assertEqual( + self._get_stored(name, "org.example.foo"), stored_content + ) + + def test_empty_expect_with_stored_revision(self) -> None: + """An empty expected revision fails against an existing stored revision.""" + for name, path in self.endpoints: + with self.subTest(endpoint=name): + self._put(path, "org.example.foo", {"com.beeper.revision_id": "abc"}) + + channel = self._put( + path, "org.example.foo", {"v": 2}, expect_revision_id="" + ) + self.assertEqual(channel.code, 409, channel.result) + self.assertEqual( + channel.json_body["errcode"], Codes.EXPECTED_REVISION_ID_MISMATCH + ) diff --git a/tests/storage/test_account_data.py b/tests/storage/test_account_data.py index 72549406a..73779d31e 100644 --- a/tests/storage/test_account_data.py +++ b/tests/storage/test_account_data.py @@ -19,13 +19,17 @@ # # -from typing import Iterable +import json +from typing import Any, Callable, Iterable, Mapping +from unittest.mock import patch from twisted.internet.testing import MemoryReactor from synapse.api.constants import AccountDataTypes from synapse.api.errors import Codes, SynapseError from synapse.server import HomeServer +from synapse.storage.database import LoggingTransaction +from synapse.types import JsonDict from synapse.util.clock import Clock from tests import unittest @@ -200,3 +204,140 @@ def get_latest_ignore_streampos(user_id: str) -> int | None: self._update_ignore_list("@foo:test", "@another:remote") self.assertEqual(get_latest_ignore_streampos("@user:test"), 3) + + +class RoomAccountDataCASTestCase(unittest.HomeserverTestCase): + """Tests the compare-and-swap path of add_account_data_to_room, which was + converted from an autocommit upsert to a transaction for this feature.""" + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.store = self.hs.get_datastores().main + self.user = "@user:test" + self.room = "!room:test" + self.data_type = "org.example.foo" + + def _get_stored(self) -> dict | None: + content = self.get_success( + self.store.get_account_data_for_room_and_type( + self.user, self.room, self.data_type + ) + ) + return dict(content) if content is not None else None + + def test_cas_match_writes_and_caches(self) -> None: + self.get_success( + self.store.add_account_data_to_room( + self.user, + self.room, + self.data_type, + {"com.beeper.revision_id": "abc", "v": 1}, + ) + ) + + new_content = {"com.beeper.revision_id": "def", "v": 2} + self.get_success( + self.store.add_account_data_to_room( + self.user, + self.room, + self.data_type, + new_content, + expected_revision_id="abc", + ) + ) + self.assertEqual(self._get_stored(), new_content) + + def test_cas_mismatch_raises_and_leaves_data_unchanged(self) -> None: + stored_content = {"com.beeper.revision_id": "abc", "v": 1} + self.get_success( + self.store.add_account_data_to_room( + self.user, self.room, self.data_type, stored_content + ) + ) + + f = self.get_failure( + self.store.add_account_data_to_room( + self.user, + self.room, + self.data_type, + {"v": 2}, + expected_revision_id="xyz", + ), + SynapseError, + ).value + self.assertEqual(f.code, 409) + self.assertEqual(f.errcode, Codes.EXPECTED_REVISION_ID_MISMATCH) + self.assertEqual( + f.error_dict(None)["com.beeper.current_content"], stored_content + ) + + self.assertEqual(self._get_stored(), stored_content) + + def _racing_insert(self, competitor_content: JsonDict) -> Callable[..., bool]: + """Returns a wrapper for the native upsert used by the CAS no-row + branch that inserts a competitor row just before the first insert + runs, simulating a concurrent first write winning the race.""" + real_insert = self.store.db_pool.simple_upsert_txn_native_upsert + raced = False + + def racing_insert( + txn: LoggingTransaction, + table: str, + keyvalues: Mapping[str, Any], + values: Mapping[str, Any], + insertion_values: Mapping[str, Any] | None = None, + where_clause: str | None = None, + ) -> bool: + nonlocal raced + if not raced: + raced = True + assert insertion_values is not None + self.store.db_pool.simple_insert_txn( + txn, + table, + { + **keyvalues, + "stream_id": insertion_values["stream_id"], + "content": json.dumps(competitor_content), + }, + ) + return real_insert( + txn, table, keyvalues, values, insertion_values, where_clause + ) + + return racing_insert + + def test_cas_first_write_race_mismatch(self) -> None: + """A concurrent first write landing between the CAS existence check + and our insert must be re-checked against the winning row.""" + competitor_content = {"com.beeper.revision_id": "theirs", "v": 1} + with patch.object( + self.store.db_pool, + "simple_upsert_txn_native_upsert", + self._racing_insert(competitor_content), + ): + f = self.get_failure( + self.store.add_account_data_to_room( + self.user, self.room, self.data_type, {"v": 2}, "mine" + ), + SynapseError, + ).value + self.assertEqual(f.code, 409) + self.assertEqual(f.errcode, Codes.EXPECTED_REVISION_ID_MISMATCH) + self.assertEqual( + f.error_dict(None)["com.beeper.current_content"], competitor_content + ) + + def test_cas_first_write_race_match(self) -> None: + """If the concurrently-written row has no revision id, our expected + revision still matches it and the write goes through.""" + with patch.object( + self.store.db_pool, + "simple_upsert_txn_native_upsert", + self._racing_insert({"v": 1}), + ): + self.get_success( + self.store.add_account_data_to_room( + self.user, self.room, self.data_type, {"v": 2}, "mine" + ) + ) + self.assertEqual(self._get_stored(), {"v": 2}) From c303e9601b0298f3bb53edf645abd960a206a35d Mon Sep 17 00:00:00 2001 From: SpiritCroc Date: Fri, 21 Aug 2026 16:49:39 +0200 Subject: [PATCH 3/3] Revert "Claude's tests" This reverts commit ee7c381ea8a3dbb61e8901b16b40cfbc0a742555. --- tests/rest/client/test_account_data.py | 173 ------------------------- tests/storage/test_account_data.py | 143 +------------------- 2 files changed, 1 insertion(+), 315 deletions(-) diff --git a/tests/rest/client/test_account_data.py b/tests/rest/client/test_account_data.py index 73d19115f..aff873d87 100644 --- a/tests/rest/client/test_account_data.py +++ b/tests/rest/client/test_account_data.py @@ -18,21 +18,13 @@ # [This file includes modifications made by New Vector Limited] # # -import urllib.parse from unittest.mock import AsyncMock -from twisted.internet.testing import MemoryReactor - from synapse.api.constants import ReceiptTypes -from synapse.api.errors import Codes from synapse.rest import admin from synapse.rest.client import account_data, login, room -from synapse.server import HomeServer -from synapse.types import JsonDict -from synapse.util.clock import Clock from tests import unittest -from tests.server import FakeChannel class AccountDataTestCase(unittest.HomeserverTestCase): @@ -220,168 +212,3 @@ def test_beeper_inbox_state_endpoint_can_set_read_marker(self) -> None: ) ) self.assertNotEqual(existing_read_marker, new_read_marker) - - -class AccountDataCASTestCase(unittest.HomeserverTestCase): - """Tests for the com.beeper.expect_revision_id compare-and-swap query param.""" - - servlets = [ - admin.register_servlets, - login.register_servlets, - room.register_servlets, - account_data.register_servlets, - ] - - def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: - self.store = hs.get_datastores().main - self.user_id = self.register_user("user", "password") - self.tok = self.login("user", "password") - self.room_id = self.helper.create_room_as(self.user_id, tok=self.tok) - - # (name, PUT path) for both flavors of account data. - self.endpoints = [ - ("global", f"/user/{self.user_id}/account_data"), - ("room", f"/user/{self.user_id}/rooms/{self.room_id}/account_data"), - ] - - def _put( - self, - base_path: str, - account_data_type: str, - content: JsonDict, - expect_revision_id: str | None = None, - ) -> FakeChannel: - url = f"{base_path}/{account_data_type}" - if expect_revision_id is not None: - url += "?com.beeper.expect_revision_id=" + urllib.parse.quote( - expect_revision_id - ) - return self.make_request("PUT", url, content, access_token=self.tok) - - def _get_stored(self, name: str, account_data_type: str) -> JsonDict | None: - if name == "global": - content = self.get_success( - self.store.get_global_account_data_by_type_for_user( - self.user_id, account_data_type - ) - ) - else: - content = self.get_success( - self.store.get_account_data_for_room_and_type( - self.user_id, self.room_id, account_data_type - ) - ) - return dict(content) if content is not None else None - - def test_no_param_always_writes(self) -> None: - """Without the query param, writes succeed regardless of stored revision.""" - for name, path in self.endpoints: - with self.subTest(endpoint=name): - channel = self._put( - path, "org.example.foo", {"com.beeper.revision_id": "abc"} - ) - self.assertEqual(channel.code, 200, channel.result) - - channel = self._put(path, "org.example.foo", {"bar": "baz"}) - self.assertEqual(channel.code, 200, channel.result) - self.assertEqual( - self._get_stored(name, "org.example.foo"), {"bar": "baz"} - ) - - def test_expect_with_no_existing_data(self) -> None: - """Any expected revision (including empty) matches when no data exists.""" - for name, path in self.endpoints: - with self.subTest(endpoint=name): - channel = self._put( - path, "org.example.new1", {"a": 1}, expect_revision_id="anything" - ) - self.assertEqual(channel.code, 200, channel.result) - - channel = self._put( - path, "org.example.new2", {"a": 1}, expect_revision_id="" - ) - self.assertEqual(channel.code, 200, channel.result) - - def test_expect_with_no_stored_revision(self) -> None: - """Any expected revision matches when stored content lacks a revision id.""" - for name, path in self.endpoints: - with self.subTest(endpoint=name): - self._put(path, "org.example.foo", {"bar": "baz"}) - - channel = self._put( - path, "org.example.foo", {"a": 1}, expect_revision_id="xyz" - ) - self.assertEqual(channel.code, 200, channel.result) - - def test_expect_with_non_string_stored_revision(self) -> None: - """A non-string stored revision id is treated as unset.""" - for name, path in self.endpoints: - with self.subTest(endpoint=name): - self._put(path, "org.example.foo", {"com.beeper.revision_id": 5}) - - channel = self._put( - path, "org.example.foo", {"a": 1}, expect_revision_id="xyz" - ) - self.assertEqual(channel.code, 200, channel.result) - - def test_expect_match(self) -> None: - """A matching expected revision allows the write; new content need not - carry a revision id itself.""" - for name, path in self.endpoints: - with self.subTest(endpoint=name): - self._put( - path, - "org.example.foo", - {"com.beeper.revision_id": "abc", "v": 1}, - ) - - new_content = {"com.beeper.revision_id": "def", "v": 2} - channel = self._put( - path, "org.example.foo", new_content, expect_revision_id="abc" - ) - self.assertEqual(channel.code, 200, channel.result) - self.assertEqual(self._get_stored(name, "org.example.foo"), new_content) - - # A revision id in the new content is not required. - channel = self._put( - path, "org.example.foo", {"v": 3}, expect_revision_id="def" - ) - self.assertEqual(channel.code, 200, channel.result) - self.assertEqual(self._get_stored(name, "org.example.foo"), {"v": 3}) - - def test_expect_mismatch(self) -> None: - """A mismatched expected revision is rejected with 409 and the stored - content is returned in the error body.""" - stored_content = {"com.beeper.revision_id": "abc", "v": 1} - for name, path in self.endpoints: - with self.subTest(endpoint=name): - self._put(path, "org.example.foo", stored_content) - - channel = self._put( - path, "org.example.foo", {"v": 2}, expect_revision_id="xyz" - ) - self.assertEqual(channel.code, 409, channel.result) - self.assertEqual( - channel.json_body["errcode"], Codes.EXPECTED_REVISION_ID_MISMATCH - ) - self.assertEqual( - channel.json_body["com.beeper.current_content"], stored_content - ) - # The stored data is unchanged. - self.assertEqual( - self._get_stored(name, "org.example.foo"), stored_content - ) - - def test_empty_expect_with_stored_revision(self) -> None: - """An empty expected revision fails against an existing stored revision.""" - for name, path in self.endpoints: - with self.subTest(endpoint=name): - self._put(path, "org.example.foo", {"com.beeper.revision_id": "abc"}) - - channel = self._put( - path, "org.example.foo", {"v": 2}, expect_revision_id="" - ) - self.assertEqual(channel.code, 409, channel.result) - self.assertEqual( - channel.json_body["errcode"], Codes.EXPECTED_REVISION_ID_MISMATCH - ) diff --git a/tests/storage/test_account_data.py b/tests/storage/test_account_data.py index 73779d31e..72549406a 100644 --- a/tests/storage/test_account_data.py +++ b/tests/storage/test_account_data.py @@ -19,17 +19,13 @@ # # -import json -from typing import Any, Callable, Iterable, Mapping -from unittest.mock import patch +from typing import Iterable from twisted.internet.testing import MemoryReactor from synapse.api.constants import AccountDataTypes from synapse.api.errors import Codes, SynapseError from synapse.server import HomeServer -from synapse.storage.database import LoggingTransaction -from synapse.types import JsonDict from synapse.util.clock import Clock from tests import unittest @@ -204,140 +200,3 @@ def get_latest_ignore_streampos(user_id: str) -> int | None: self._update_ignore_list("@foo:test", "@another:remote") self.assertEqual(get_latest_ignore_streampos("@user:test"), 3) - - -class RoomAccountDataCASTestCase(unittest.HomeserverTestCase): - """Tests the compare-and-swap path of add_account_data_to_room, which was - converted from an autocommit upsert to a transaction for this feature.""" - - def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: - self.store = self.hs.get_datastores().main - self.user = "@user:test" - self.room = "!room:test" - self.data_type = "org.example.foo" - - def _get_stored(self) -> dict | None: - content = self.get_success( - self.store.get_account_data_for_room_and_type( - self.user, self.room, self.data_type - ) - ) - return dict(content) if content is not None else None - - def test_cas_match_writes_and_caches(self) -> None: - self.get_success( - self.store.add_account_data_to_room( - self.user, - self.room, - self.data_type, - {"com.beeper.revision_id": "abc", "v": 1}, - ) - ) - - new_content = {"com.beeper.revision_id": "def", "v": 2} - self.get_success( - self.store.add_account_data_to_room( - self.user, - self.room, - self.data_type, - new_content, - expected_revision_id="abc", - ) - ) - self.assertEqual(self._get_stored(), new_content) - - def test_cas_mismatch_raises_and_leaves_data_unchanged(self) -> None: - stored_content = {"com.beeper.revision_id": "abc", "v": 1} - self.get_success( - self.store.add_account_data_to_room( - self.user, self.room, self.data_type, stored_content - ) - ) - - f = self.get_failure( - self.store.add_account_data_to_room( - self.user, - self.room, - self.data_type, - {"v": 2}, - expected_revision_id="xyz", - ), - SynapseError, - ).value - self.assertEqual(f.code, 409) - self.assertEqual(f.errcode, Codes.EXPECTED_REVISION_ID_MISMATCH) - self.assertEqual( - f.error_dict(None)["com.beeper.current_content"], stored_content - ) - - self.assertEqual(self._get_stored(), stored_content) - - def _racing_insert(self, competitor_content: JsonDict) -> Callable[..., bool]: - """Returns a wrapper for the native upsert used by the CAS no-row - branch that inserts a competitor row just before the first insert - runs, simulating a concurrent first write winning the race.""" - real_insert = self.store.db_pool.simple_upsert_txn_native_upsert - raced = False - - def racing_insert( - txn: LoggingTransaction, - table: str, - keyvalues: Mapping[str, Any], - values: Mapping[str, Any], - insertion_values: Mapping[str, Any] | None = None, - where_clause: str | None = None, - ) -> bool: - nonlocal raced - if not raced: - raced = True - assert insertion_values is not None - self.store.db_pool.simple_insert_txn( - txn, - table, - { - **keyvalues, - "stream_id": insertion_values["stream_id"], - "content": json.dumps(competitor_content), - }, - ) - return real_insert( - txn, table, keyvalues, values, insertion_values, where_clause - ) - - return racing_insert - - def test_cas_first_write_race_mismatch(self) -> None: - """A concurrent first write landing between the CAS existence check - and our insert must be re-checked against the winning row.""" - competitor_content = {"com.beeper.revision_id": "theirs", "v": 1} - with patch.object( - self.store.db_pool, - "simple_upsert_txn_native_upsert", - self._racing_insert(competitor_content), - ): - f = self.get_failure( - self.store.add_account_data_to_room( - self.user, self.room, self.data_type, {"v": 2}, "mine" - ), - SynapseError, - ).value - self.assertEqual(f.code, 409) - self.assertEqual(f.errcode, Codes.EXPECTED_REVISION_ID_MISMATCH) - self.assertEqual( - f.error_dict(None)["com.beeper.current_content"], competitor_content - ) - - def test_cas_first_write_race_match(self) -> None: - """If the concurrently-written row has no revision id, our expected - revision still matches it and the write goes through.""" - with patch.object( - self.store.db_pool, - "simple_upsert_txn_native_upsert", - self._racing_insert({"v": 1}), - ): - self.get_success( - self.store.add_account_data_to_room( - self.user, self.room, self.data_type, {"v": 2}, "mine" - ) - ) - self.assertEqual(self._get_stored(), {"v": 2})