test_scheduler.py 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425
  1. # Copyright 2015, 2016 OpenMarket Ltd
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. from typing import List, Optional, Sequence, Tuple, cast
  15. from unittest.mock import AsyncMock, Mock
  16. from typing_extensions import TypeAlias
  17. from twisted.internet import defer
  18. from twisted.test.proto_helpers import MemoryReactor
  19. from synapse.appservice import (
  20. ApplicationService,
  21. ApplicationServiceState,
  22. TransactionOneTimeKeysCount,
  23. TransactionUnusedFallbackKeys,
  24. )
  25. from synapse.appservice.scheduler import (
  26. ApplicationServiceScheduler,
  27. _Recoverer,
  28. _TransactionController,
  29. )
  30. from synapse.events import EventBase
  31. from synapse.logging.context import make_deferred_yieldable
  32. from synapse.server import HomeServer
  33. from synapse.types import DeviceListUpdates, JsonDict
  34. from synapse.util import Clock
  35. from tests import unittest
  36. from ..utils import MockClock
  37. class ApplicationServiceSchedulerTransactionCtrlTestCase(unittest.TestCase):
  38. def setUp(self) -> None:
  39. self.clock = MockClock()
  40. self.store = Mock()
  41. self.as_api = Mock()
  42. self.recoverer = Mock()
  43. self.recoverer_fn = Mock(return_value=self.recoverer)
  44. self.txnctrl = _TransactionController(
  45. clock=cast(Clock, self.clock), store=self.store, as_api=self.as_api
  46. )
  47. self.txnctrl.RECOVERER_CLASS = self.recoverer_fn
  48. def test_single_service_up_txn_sent(self) -> None:
  49. # Test: The AS is up and the txn is successfully sent.
  50. service = Mock()
  51. events = [Mock(), Mock()]
  52. txn_id = "foobar"
  53. txn = Mock(id=txn_id, service=service, events=events)
  54. # mock methods
  55. self.store.get_appservice_state = AsyncMock(
  56. return_value=ApplicationServiceState.UP
  57. )
  58. txn.send = AsyncMock(return_value=True)
  59. txn.complete = AsyncMock(return_value=True)
  60. self.store.create_appservice_txn = AsyncMock(return_value=txn)
  61. # actual call
  62. self.successResultOf(defer.ensureDeferred(self.txnctrl.send(service, events)))
  63. self.store.create_appservice_txn.assert_called_once_with(
  64. service=service,
  65. events=events,
  66. ephemeral=[],
  67. to_device_messages=[], # txn made and saved
  68. one_time_keys_count={},
  69. unused_fallback_keys={},
  70. device_list_summary=DeviceListUpdates(),
  71. )
  72. self.assertEqual(0, len(self.txnctrl.recoverers)) # no recoverer made
  73. txn.complete.assert_called_once_with(self.store) # txn completed
  74. def test_single_service_down(self) -> None:
  75. # Test: The AS is down so it shouldn't push; Recoverers will do it.
  76. # It should still make a transaction though.
  77. service = Mock()
  78. events = [Mock(), Mock()]
  79. txn = Mock(id="idhere", service=service, events=events)
  80. self.store.get_appservice_state = AsyncMock(
  81. return_value=ApplicationServiceState.DOWN
  82. )
  83. self.store.create_appservice_txn = AsyncMock(return_value=txn)
  84. # actual call
  85. self.successResultOf(defer.ensureDeferred(self.txnctrl.send(service, events)))
  86. self.store.create_appservice_txn.assert_called_once_with(
  87. service=service,
  88. events=events,
  89. ephemeral=[],
  90. to_device_messages=[], # txn made and saved
  91. one_time_keys_count={},
  92. unused_fallback_keys={},
  93. device_list_summary=DeviceListUpdates(),
  94. )
  95. self.assertEqual(0, txn.send.call_count) # txn not sent though
  96. self.assertEqual(0, txn.complete.call_count) # or completed
  97. def test_single_service_up_txn_not_sent(self) -> None:
  98. # Test: The AS is up and the txn is not sent. A Recoverer is made and
  99. # started.
  100. service = Mock()
  101. events = [Mock(), Mock()]
  102. txn_id = "foobar"
  103. txn = Mock(id=txn_id, service=service, events=events)
  104. # mock methods
  105. self.store.get_appservice_state = AsyncMock(
  106. return_value=ApplicationServiceState.UP
  107. )
  108. self.store.set_appservice_state = AsyncMock(return_value=True)
  109. txn.send = AsyncMock(return_value=False) # fails to send
  110. self.store.create_appservice_txn = AsyncMock(return_value=txn)
  111. # actual call
  112. self.successResultOf(defer.ensureDeferred(self.txnctrl.send(service, events)))
  113. self.store.create_appservice_txn.assert_called_once_with(
  114. service=service,
  115. events=events,
  116. ephemeral=[],
  117. to_device_messages=[],
  118. one_time_keys_count={},
  119. unused_fallback_keys={},
  120. device_list_summary=DeviceListUpdates(),
  121. )
  122. self.assertEqual(1, self.recoverer_fn.call_count) # recoverer made
  123. self.assertEqual(1, self.recoverer.recover.call_count) # and invoked
  124. self.assertEqual(1, len(self.txnctrl.recoverers)) # and stored
  125. self.assertEqual(0, txn.complete.call_count) # txn not completed
  126. self.store.set_appservice_state.assert_called_once_with(
  127. service, ApplicationServiceState.DOWN # service marked as down
  128. )
  129. class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase):
  130. def setUp(self) -> None:
  131. self.clock = MockClock()
  132. self.as_api = Mock()
  133. self.store = Mock()
  134. self.service = Mock()
  135. self.callback = AsyncMock()
  136. self.recoverer = _Recoverer(
  137. clock=cast(Clock, self.clock),
  138. as_api=self.as_api,
  139. store=self.store,
  140. service=self.service,
  141. callback=self.callback,
  142. )
  143. def test_recover_single_txn(self) -> None:
  144. txn = Mock()
  145. # return one txn to send, then no more old txns
  146. txns = [txn, None]
  147. def take_txn(
  148. *args: object, **kwargs: object
  149. ) -> "defer.Deferred[Optional[Mock]]":
  150. return defer.succeed(txns.pop(0))
  151. self.store.get_oldest_unsent_txn = Mock(side_effect=take_txn)
  152. self.recoverer.recover()
  153. # shouldn't have called anything prior to waiting for exp backoff
  154. self.assertEqual(0, self.store.get_oldest_unsent_txn.call_count)
  155. txn.send = AsyncMock(return_value=True)
  156. txn.complete = AsyncMock(return_value=None)
  157. # wait for exp backoff
  158. self.clock.advance_time(2)
  159. self.assertEqual(1, txn.send.call_count)
  160. self.assertEqual(1, txn.complete.call_count)
  161. # 2 because it needs to get None to know there are no more txns
  162. self.assertEqual(2, self.store.get_oldest_unsent_txn.call_count)
  163. self.callback.assert_called_once_with(self.recoverer)
  164. self.assertEqual(self.recoverer.service, self.service)
  165. def test_recover_retry_txn(self) -> None:
  166. txn = Mock()
  167. txns = [txn, None]
  168. pop_txn = False
  169. def take_txn(
  170. *args: object, **kwargs: object
  171. ) -> "defer.Deferred[Optional[Mock]]":
  172. if pop_txn:
  173. return defer.succeed(txns.pop(0))
  174. else:
  175. return defer.succeed(txn)
  176. self.store.get_oldest_unsent_txn = Mock(side_effect=take_txn)
  177. self.recoverer.recover()
  178. self.assertEqual(0, self.store.get_oldest_unsent_txn.call_count)
  179. txn.send = AsyncMock(return_value=False)
  180. txn.complete = AsyncMock(return_value=None)
  181. self.clock.advance_time(2)
  182. self.assertEqual(1, txn.send.call_count)
  183. self.assertEqual(0, txn.complete.call_count)
  184. self.assertEqual(0, self.callback.call_count)
  185. self.clock.advance_time(4)
  186. self.assertEqual(2, txn.send.call_count)
  187. self.assertEqual(0, txn.complete.call_count)
  188. self.assertEqual(0, self.callback.call_count)
  189. self.clock.advance_time(8)
  190. self.assertEqual(3, txn.send.call_count)
  191. self.assertEqual(0, txn.complete.call_count)
  192. self.assertEqual(0, self.callback.call_count)
  193. txn.send = AsyncMock(return_value=True) # successfully send the txn
  194. pop_txn = True # returns the txn the first time, then no more.
  195. self.clock.advance_time(16)
  196. self.assertEqual(1, txn.send.call_count) # new mock reset call count
  197. self.assertEqual(1, txn.complete.call_count)
  198. self.callback.assert_called_once_with(self.recoverer)
  199. # Corresponds to synapse.appservice.scheduler._TransactionController.send
  200. TxnCtrlArgs: TypeAlias = """
  201. defer.Deferred[
  202. Tuple[
  203. ApplicationService,
  204. Sequence[EventBase],
  205. Optional[List[JsonDict]],
  206. Optional[List[JsonDict]],
  207. Optional[TransactionOneTimeKeysCount],
  208. Optional[TransactionUnusedFallbackKeys],
  209. Optional[DeviceListUpdates],
  210. ]
  211. ]
  212. """
  213. class ApplicationServiceSchedulerQueuerTestCase(unittest.HomeserverTestCase):
  214. def prepare(self, reactor: "MemoryReactor", clock: Clock, hs: HomeServer) -> None:
  215. self.scheduler = ApplicationServiceScheduler(hs)
  216. self.txn_ctrl = Mock()
  217. self.txn_ctrl.send = AsyncMock()
  218. # Replace instantiated _TransactionController instances with our Mock
  219. self.scheduler.txn_ctrl = self.txn_ctrl
  220. self.scheduler.queuer.txn_ctrl = self.txn_ctrl
  221. def test_send_single_event_no_queue(self) -> None:
  222. # Expect the event to be sent immediately.
  223. service = Mock(id=4)
  224. event = Mock()
  225. self.scheduler.enqueue_for_appservice(service, events=[event])
  226. self.txn_ctrl.send.assert_called_once_with(
  227. service, [event], [], [], None, None, DeviceListUpdates()
  228. )
  229. def test_send_single_event_with_queue(self) -> None:
  230. d: TxnCtrlArgs = defer.Deferred()
  231. self.txn_ctrl.send = Mock(return_value=make_deferred_yieldable(d))
  232. service = Mock(id=4)
  233. event = Mock(event_id="first")
  234. event2 = Mock(event_id="second")
  235. event3 = Mock(event_id="third")
  236. # Send an event and don't resolve it just yet.
  237. self.scheduler.enqueue_for_appservice(service, events=[event])
  238. # Send more events: expect send() to NOT be called multiple times.
  239. # (call enqueue_for_appservice multiple times deliberately)
  240. self.scheduler.enqueue_for_appservice(service, events=[event2])
  241. self.scheduler.enqueue_for_appservice(service, events=[event3])
  242. self.txn_ctrl.send.assert_called_with(
  243. service, [event], [], [], None, None, DeviceListUpdates()
  244. )
  245. self.assertEqual(1, self.txn_ctrl.send.call_count)
  246. # Resolve the send event: expect the queued events to be sent
  247. d.callback(service)
  248. self.txn_ctrl.send.assert_called_with(
  249. service, [event2, event3], [], [], None, None, DeviceListUpdates()
  250. )
  251. self.assertEqual(2, self.txn_ctrl.send.call_count)
  252. def test_multiple_service_queues(self) -> None:
  253. # Tests that each service has its own queue, and that they don't block
  254. # on each other.
  255. srv1 = Mock(id=4)
  256. srv_1_defer: "defer.Deferred[EventBase]" = defer.Deferred()
  257. srv_1_event = Mock(event_id="srv1a")
  258. srv_1_event2 = Mock(event_id="srv1b")
  259. srv2 = Mock(id=6)
  260. srv_2_defer: "defer.Deferred[EventBase]" = defer.Deferred()
  261. srv_2_event = Mock(event_id="srv2a")
  262. srv_2_event2 = Mock(event_id="srv2b")
  263. send_return_list = [srv_1_defer, srv_2_defer]
  264. def do_send(*args: object, **kwargs: object) -> "defer.Deferred[EventBase]":
  265. return make_deferred_yieldable(send_return_list.pop(0))
  266. self.txn_ctrl.send = Mock(side_effect=do_send)
  267. # send events for different ASes and make sure they are sent
  268. self.scheduler.enqueue_for_appservice(srv1, events=[srv_1_event])
  269. self.scheduler.enqueue_for_appservice(srv1, events=[srv_1_event2])
  270. self.txn_ctrl.send.assert_called_with(
  271. srv1, [srv_1_event], [], [], None, None, DeviceListUpdates()
  272. )
  273. self.scheduler.enqueue_for_appservice(srv2, events=[srv_2_event])
  274. self.scheduler.enqueue_for_appservice(srv2, events=[srv_2_event2])
  275. self.txn_ctrl.send.assert_called_with(
  276. srv2, [srv_2_event], [], [], None, None, DeviceListUpdates()
  277. )
  278. # make sure callbacks for a service only send queued events for THAT
  279. # service
  280. srv_2_defer.callback(srv2)
  281. self.txn_ctrl.send.assert_called_with(
  282. srv2, [srv_2_event2], [], [], None, None, DeviceListUpdates()
  283. )
  284. self.assertEqual(3, self.txn_ctrl.send.call_count)
  285. def test_send_large_txns(self) -> None:
  286. srv_1_defer: "defer.Deferred[EventBase]" = defer.Deferred()
  287. srv_2_defer: "defer.Deferred[EventBase]" = defer.Deferred()
  288. send_return_list = [srv_1_defer, srv_2_defer]
  289. def do_send(*args: object, **kwargs: object) -> "defer.Deferred[EventBase]":
  290. return make_deferred_yieldable(send_return_list.pop(0))
  291. self.txn_ctrl.send = Mock(side_effect=do_send)
  292. service = Mock(id=4, name="service")
  293. event_list = [Mock(name="event%i" % (i + 1)) for i in range(200)]
  294. for event in event_list:
  295. self.scheduler.enqueue_for_appservice(service, [event], [])
  296. # Expect the first event to be sent immediately.
  297. self.txn_ctrl.send.assert_called_with(
  298. service, [event_list[0]], [], [], None, None, DeviceListUpdates()
  299. )
  300. srv_1_defer.callback(service)
  301. # Then send the next 100 events
  302. self.txn_ctrl.send.assert_called_with(
  303. service, event_list[1:101], [], [], None, None, DeviceListUpdates()
  304. )
  305. srv_2_defer.callback(service)
  306. # Then the final 99 events
  307. self.txn_ctrl.send.assert_called_with(
  308. service, event_list[101:], [], [], None, None, DeviceListUpdates()
  309. )
  310. self.assertEqual(3, self.txn_ctrl.send.call_count)
  311. def test_send_single_ephemeral_no_queue(self) -> None:
  312. # Expect the event to be sent immediately.
  313. service = Mock(id=4, name="service")
  314. event_list = [Mock(name="event")]
  315. self.scheduler.enqueue_for_appservice(service, ephemeral=event_list)
  316. self.txn_ctrl.send.assert_called_once_with(
  317. service, [], event_list, [], None, None, DeviceListUpdates()
  318. )
  319. def test_send_multiple_ephemeral_no_queue(self) -> None:
  320. # Expect the event to be sent immediately.
  321. service = Mock(id=4, name="service")
  322. event_list = [Mock(name="event1"), Mock(name="event2"), Mock(name="event3")]
  323. self.scheduler.enqueue_for_appservice(service, ephemeral=event_list)
  324. self.txn_ctrl.send.assert_called_once_with(
  325. service, [], event_list, [], None, None, DeviceListUpdates()
  326. )
  327. def test_send_single_ephemeral_with_queue(self) -> None:
  328. d: TxnCtrlArgs = defer.Deferred()
  329. self.txn_ctrl.send = Mock(return_value=make_deferred_yieldable(d))
  330. service = Mock(id=4)
  331. event_list_1 = [Mock(event_id="event1"), Mock(event_id="event2")]
  332. event_list_2 = [Mock(event_id="event3"), Mock(event_id="event4")]
  333. event_list_3 = [Mock(event_id="event5"), Mock(event_id="event6")]
  334. # Send an event and don't resolve it just yet.
  335. self.scheduler.enqueue_for_appservice(service, ephemeral=event_list_1)
  336. # Send more events: expect send() to NOT be called multiple times.
  337. self.scheduler.enqueue_for_appservice(service, ephemeral=event_list_2)
  338. self.scheduler.enqueue_for_appservice(service, ephemeral=event_list_3)
  339. self.txn_ctrl.send.assert_called_with(
  340. service, [], event_list_1, [], None, None, DeviceListUpdates()
  341. )
  342. self.assertEqual(1, self.txn_ctrl.send.call_count)
  343. # Resolve txn_ctrl.send
  344. d.callback(service)
  345. # Expect the queued events to be sent
  346. self.txn_ctrl.send.assert_called_with(
  347. service,
  348. [],
  349. event_list_2 + event_list_3,
  350. [],
  351. None,
  352. None,
  353. DeviceListUpdates(),
  354. )
  355. self.assertEqual(2, self.txn_ctrl.send.call_count)
  356. def test_send_large_txns_ephemeral(self) -> None:
  357. d: TxnCtrlArgs = defer.Deferred()
  358. self.txn_ctrl.send = Mock(return_value=make_deferred_yieldable(d))
  359. # Expect the event to be sent immediately.
  360. service = Mock(id=4, name="service")
  361. first_chunk = [Mock(name="event%i" % (i + 1)) for i in range(100)]
  362. second_chunk = [Mock(name="event%i" % (i + 101)) for i in range(50)]
  363. event_list = first_chunk + second_chunk
  364. self.scheduler.enqueue_for_appservice(service, ephemeral=event_list)
  365. self.txn_ctrl.send.assert_called_once_with(
  366. service, [], first_chunk, [], None, None, DeviceListUpdates()
  367. )
  368. d.callback(service)
  369. self.txn_ctrl.send.assert_called_with(
  370. service, [], second_chunk, [], None, None, DeviceListUpdates()
  371. )
  372. self.assertEqual(2, self.txn_ctrl.send.call_count)