1
0

test_scheduler.py 14 KB

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