test_scheduler.py 16 KB

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