mirrors/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2024-11-25 11:05:49 +03:00
Code Issues Projects Releases Packages Wiki Activity

Fix get|set_type_stream_id_for_appservice store functions (#8648)

Browse source
This commit is contained in:
Will Hunt 2020-10-26 14:51:33 +00:00 committed by GitHub
parent 73d8209694
commit e8dbbcb64c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 85 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
changelog.d/8648.bugfix Normal file
View file

@ -0,0 +1 @@
Fix a bug introduced in v1.22.0rc1 which would cause ephemeral events to not be sent to appservices.

12
synapse/handlers/appservice.py
View file

@ -236,16 +236,16 @@ class ApplicationServicesHandler:
events = await self._handle_receipts(service)
if events:
self.scheduler.submit_ephemeral_events_for_as(service, events)
await self.store.set_type_stream_id_for_appservice(
service, "read_receipt", new_token
)
await self.store.set_type_stream_id_for_appservice(
service, "read_receipt", new_token
)
elif stream_key == "presence_key":
events = await self._handle_presence(service, users)
if events:
self.scheduler.submit_ephemeral_events_for_as(service, events)
await self.store.set_type_stream_id_for_appservice(
service, "presence", new_token
)
await self.store.set_type_stream_id_for_appservice(
service, "presence", new_token
)
async def _handle_typing(self, service: ApplicationService, new_token: int):
typing_source = self.event_sources.sources["typing"]

29
synapse/storage/databases/main/appservice.py
View file

@ -369,17 +369,25 @@ class ApplicationServiceTransactionWorkerStore(
async def get_type_stream_id_for_appservice(
self, service: ApplicationService, type: str
) -> int:
if type not in ("read_receipt", "presence"):
raise ValueError(
"Expected type to be a valid application stream id type, got %s"
% (type,)
)
def get_type_stream_id_for_appservice_txn(txn):
stream_id_type = "%s_stream_id" % type
txn.execute(
"SELECT ? FROM application_services_state WHERE as_id=?",
(stream_id_type, service.id,),
# We do NOT want to escape `stream_id_type`.
"SELECT %s FROM application_services_state WHERE as_id=?"
% stream_id_type,
(service.id,),
)
last_txn_id = txn.fetchone()
if last_txn_id is None or last_txn_id[0] is None: # no row exists
last_stream_id = txn.fetchone()
if last_stream_id is None or last_stream_id[0] is None: # no row exists
return 0
else:
return int(last_txn_id[0])
return int(last_stream_id[0])
return await self.db_pool.runInteraction(
"get_type_stream_id_for_appservice", get_type_stream_id_for_appservice_txn
@ -388,11 +396,18 @@ class ApplicationServiceTransactionWorkerStore(
async def set_type_stream_id_for_appservice(
self, service: ApplicationService, type: str, pos: int
) -> None:
if type not in ("read_receipt", "presence"):
raise ValueError(
"Expected type to be a valid application stream id type, got %s"
% (type,)
)
def set_type_stream_id_for_appservice_txn(txn):
stream_id_type = "%s_stream_id" % type
txn.execute(
"UPDATE ? SET device_list_stream_id = ? WHERE as_id=?",
(stream_id_type, pos, service.id),
"UPDATE application_services_state SET %s = ? WHERE as_id=?"
% stream_id_type,
(pos, service.id),
)
await self.db_pool.runInteraction(

56
tests/storage/test_appservice.py
View file

@ -410,6 +410,62 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase):
)
class ApplicationServiceStoreTypeStreamIds(unittest.HomeserverTestCase):
def make_homeserver(self, reactor, clock):
hs = self.setup_test_homeserver()
return hs
def prepare(self, hs, reactor, clock):
self.service = Mock(id="foo")
self.store = self.hs.get_datastore()
self.get_success(self.store.set_appservice_state(self.service, "up"))
def test_get_type_stream_id_for_appservice_no_value(self):
value = self.get_success(
self.store.get_type_stream_id_for_appservice(self.service, "read_receipt")
)
self.assertEquals(value, 0)
value = self.get_success(
self.store.get_type_stream_id_for_appservice(self.service, "presence")
)
self.assertEquals(value, 0)
def test_get_type_stream_id_for_appservice_invalid_type(self):
self.get_failure(
self.store.get_type_stream_id_for_appservice(self.service, "foobar"),
ValueError,
)
def test_set_type_stream_id_for_appservice(self):
read_receipt_value = 1024
self.get_success(
self.store.set_type_stream_id_for_appservice(
self.service, "read_receipt", read_receipt_value
)
)
result = self.get_success(
self.store.get_type_stream_id_for_appservice(self.service, "read_receipt")
)
self.assertEqual(result, read_receipt_value)
self.get_success(
self.store.set_type_stream_id_for_appservice(
self.service, "presence", read_receipt_value
)
)
result = self.get_success(
self.store.get_type_stream_id_for_appservice(self.service, "presence")
)
self.assertEqual(result, read_receipt_value)
def test_set_type_stream_id_for_appservice_invalid_type(self):
self.get_failure(
self.store.set_type_stream_id_for_appservice(self.service, "foobar", 1024),
ValueError,
)
# required for ApplicationServiceTransactionStoreTestCase tests
class TestTransactionStore(ApplicationServiceTransactionStore, ApplicationServiceStore):
def __init__(self, database: DatabasePool, db_conn, hs):