From 192e228a98f3700f48d7fd136f4dce2979ec7c90 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Fri, 6 Mar 2015 11:50:27 +0000 Subject: Start adding some tests --- tests/appservice/test_scheduler.py | 106 +++++++++++++++++++++++++++++++++++++ 1 file changed, 106 insertions(+) create mode 100644 tests/appservice/test_scheduler.py (limited to 'tests') diff --git a/tests/appservice/test_scheduler.py b/tests/appservice/test_scheduler.py new file mode 100644 index 0000000000..b41d4358cf --- /dev/null +++ b/tests/appservice/test_scheduler.py @@ -0,0 +1,106 @@ +# -*- coding: utf-8 -*- +# Copyright 2015 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from synapse.appservice.scheduler import ( + AppServiceScheduler, AppServiceTransaction, _EventGrouper, + _TransactionController, _Recoverer +) +from twisted.internet import defer +from ..utils import MockClock +from mock import Mock +from tests import unittest + +class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase): + + def setUp(self): + self.clock = MockClock() + self.as_api = Mock() + self.store = Mock() + self.service = Mock() + self.callback = Mock() + self.recoverer = _Recoverer( + clock=self.clock, + as_api=self.as_api, + store=self.store, + service=self.service, + callback=self.callback, + ) + + def test_recover_service_single_txn(self): + txns = self._mk_txns(1) + self.store.get_oldest_txn = Mock(return_value=defer.succeed(txns[0])) + + self.recoverer.recover() + self.assertEquals(0, self.store.get_oldest_txn.call_count) + self.clock.advance_time(2000) + self.assertEquals(2, self.store.get_oldest_txn.call_count) + + def _mk_txns(self, num_txns): + return [ + Mock() for i in range(num_txns) + ] + + + +class ApplicationServiceSchedulerEventGrouperTestCase(unittest.TestCase): + + def setUp(self): + self.grouper = _EventGrouper() + + def test_drain_single_event(self): + service = Mock() + event = Mock() + self.grouper.on_receive(service, event) + groups = self.grouper.drain_groups() + self.assertTrue(service in groups) + self.assertEquals([event], groups[service]) + self.assertEquals(1, len(groups.keys())) + # no more events + self.assertEquals(self.grouper.drain_groups(), {}) + + def test_drain_multiple_events(self): + service = Mock() + events = [Mock(), Mock(), Mock()] + for e in events: + self.grouper.on_receive(service, e) + groups = self.grouper.drain_groups() + self.assertTrue(service in groups) + self.assertEquals(events, groups[service]) + # no more events + self.assertEquals(self.grouper.drain_groups(), {}) + + def test_drain_multiple_services(self): + services = [Mock(), Mock(), Mock()] + events_a = [Mock(), Mock()] + events_b = [Mock()] + events_c = [Mock(), Mock(), Mock(), Mock()] + mappings = { + services[0]: events_a, + services[1]: events_b, + services[2]: events_c + } + for e in events_b: + self.grouper.on_receive(services[1], e) + for e in events_c: + self.grouper.on_receive(services[2], e) + for e in events_a: + self.grouper.on_receive(services[0], e) + + groups = self.grouper.drain_groups() + for service in services: + self.assertTrue(service in groups) + self.assertEquals(mappings[service], groups[service]) + self.assertEquals(3, len(groups.keys())) + # no more events + self.assertEquals(self.grouper.drain_groups(), {}) -- cgit 1.5.1 From 0fbfe1b08a791e95dc9e9d417f131e80b4ce8059 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Fri, 6 Mar 2015 14:36:52 +0000 Subject: Add more tests; fix bugs. --- synapse/appservice/scheduler.py | 4 +-- tests/appservice/test_scheduler.py | 54 +++++++++++++++++++++++++++++++++----- 2 files changed, 49 insertions(+), 9 deletions(-) (limited to 'tests') diff --git a/synapse/appservice/scheduler.py b/synapse/appservice/scheduler.py index 754f39381f..f54df9c9a5 100644 --- a/synapse/appservice/scheduler.py +++ b/synapse/appservice/scheduler.py @@ -174,7 +174,7 @@ class _Recoverer(object): self.backoff_counter = 1 def recover(self): - self.clock.call_later(2000 ** self.backoff_counter, self.retry) + self.clock.call_later(1000 * (2 ** self.backoff_counter), self.retry) @defer.inlineCallbacks def retry(self): @@ -184,7 +184,7 @@ class _Recoverer(object): txn.complete(self.store) # reset the backoff counter and retry immediately self.backoff_counter = 1 - self.retry() + yield self.retry() else: self.backoff_counter += 1 self.recover() diff --git a/tests/appservice/test_scheduler.py b/tests/appservice/test_scheduler.py index b41d4358cf..1e3eb9e1cc 100644 --- a/tests/appservice/test_scheduler.py +++ b/tests/appservice/test_scheduler.py @@ -21,6 +21,7 @@ from ..utils import MockClock from mock import Mock from tests import unittest + class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase): def setUp(self): @@ -37,21 +38,60 @@ class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase): callback=self.callback, ) - def test_recover_service_single_txn(self): - txns = self._mk_txns(1) - self.store.get_oldest_txn = Mock(return_value=defer.succeed(txns[0])) + def test_recover_single_txn(self): + txn = Mock() + # return one txn to send, then no more old txns + txns = [txn, None] + + def take_txn(*args, **kwargs): + return defer.succeed(txns.pop(0)) + self.store.get_oldest_txn = Mock(side_effect=take_txn) self.recoverer.recover() + # shouldn't have called anything prior to waiting for exp backoff self.assertEquals(0, self.store.get_oldest_txn.call_count) + txn.send = Mock(return_value=True) + # wait for exp backoff self.clock.advance_time(2000) + self.assertEquals(1, txn.send.call_count) + self.assertEquals(1, txn.complete.call_count) + # 2 because it needs to get None to know there are no more txns self.assertEquals(2, self.store.get_oldest_txn.call_count) + self.assertEquals(1, self.callback.call_count) - def _mk_txns(self, num_txns): - return [ - Mock() for i in range(num_txns) - ] + def test_recover_retry_txn(self): + txn = Mock() + txns = [txn, None] + pop_txn = False + def take_txn(*args, **kwargs): + if pop_txn: + return defer.succeed(txns.pop(0)) + else: + return defer.succeed(txn) + self.store.get_oldest_txn = Mock(side_effect=take_txn) + self.recoverer.recover() + self.assertEquals(0, self.store.get_oldest_txn.call_count) + txn.send = Mock(return_value=False) + self.clock.advance_time(2000) + self.assertEquals(1, txn.send.call_count) + self.assertEquals(0, txn.complete.call_count) + self.assertEquals(0, self.callback.call_count) + self.clock.advance_time(4000) + self.assertEquals(2, txn.send.call_count) + self.assertEquals(0, txn.complete.call_count) + self.assertEquals(0, self.callback.call_count) + self.clock.advance_time(8000) + self.assertEquals(3, txn.send.call_count) + self.assertEquals(0, txn.complete.call_count) + self.assertEquals(0, self.callback.call_count) + txn.send = Mock(return_value=True) # successfully send the txn + pop_txn = True # returns the txn the first time, then no more. + self.clock.advance_time(16000) + self.assertEquals(1, txn.send.call_count) # new mock reset call count + self.assertEquals(1, txn.complete.call_count) + self.assertEquals(1, self.callback.call_count) class ApplicationServiceSchedulerEventGrouperTestCase(unittest.TestCase): -- cgit 1.5.1 From f260cb72cd3435d540411962a92ca2a9fd333eb1 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Fri, 6 Mar 2015 15:12:24 +0000 Subject: Flesh out more stub functions. --- synapse/appservice/__init__.py | 5 +++++ synapse/appservice/scheduler.py | 37 +++++++++++++++++++++++++++++-------- synapse/storage/appservice.py | 17 +++++++++++++++-- tests/appservice/test_scheduler.py | 5 +++-- 4 files changed, 52 insertions(+), 12 deletions(-) (limited to 'tests') diff --git a/synapse/appservice/__init__.py b/synapse/appservice/__init__.py index a268a6bcc4..cc6c381566 100644 --- a/synapse/appservice/__init__.py +++ b/synapse/appservice/__init__.py @@ -20,6 +20,11 @@ import re logger = logging.getLogger(__name__) +class ApplicationServiceState(object): + DOWN = "down" + UP = "up" + + class ApplicationService(object): """Defines an application service. This definition is mostly what is provided to the /register AS API. diff --git a/synapse/appservice/scheduler.py b/synapse/appservice/scheduler.py index 645d7bf6b2..99e83747a8 100644 --- a/synapse/appservice/scheduler.py +++ b/synapse/appservice/scheduler.py @@ -49,7 +49,11 @@ This is all tied together by the AppServiceScheduler which DIs the required components. """ +from synapse.appservice import ApplicationServiceState from twisted.internet import defer +import logging + +logger = logging.getLogger(__name__) class AppServiceScheduler(object): @@ -162,21 +166,36 @@ class _TransactionController(object): if txn.send(self.as_api): txn.complete(self.store) else: - # TODO mark AS as down self._start_recoverer(service) self.clock.call_later(1000, self.start_polling) - def on_recovered(self, service): - # TODO mark AS as UP - pass + @defer.inlineCallbacks + def on_recovered(self, recoverer): + applied_state = yield self.store.set_appservice_state( + recoverer.service, + ApplicationServiceState.UP + ) + if not applied_state: + logger.error("Failed to apply appservice state UP to service %s", + recoverer.service) def add_recoverers(self, recoverers): for r in recoverers: self.recoverers.append(r) + @defer.inlineCallbacks def _start_recoverer(self, service): - recoverer = self.recoverer_fn(service, self.on_recovered) - recoverer.recover() + applied_state = yield self.store.set_appservice_state( + service, + ApplicationServiceState.DOWN + ) + if applied_state: + recoverer = self.recoverer_fn(service, self.on_recovered) + self.add_recoverers([recoverer]) + recoverer.recover() + else: + logger.error("Failed to apply appservice state DOWN to service %s", + service) def _is_service_up(self, service): pass @@ -193,7 +212,9 @@ class _Recoverer(object): @staticmethod @defer.inlineCallbacks def start(clock, store, as_api, callback): - services = yield store.get_failing_appservices() + services = yield store.get_appservices_by_state( + ApplicationServiceState.DOWN + ) recoverers = [ _Recoverer(clock, store, as_api, s, callback) for s in services ] @@ -228,7 +249,7 @@ class _Recoverer(object): self._set_service_recovered() def _set_service_recovered(self): - self.callback(self.service) + self.callback(self) @defer.inlineCallbacks def _get_oldest_txn(self): diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py index c1762692b9..214f6d99c5 100644 --- a/synapse/storage/appservice.py +++ b/synapse/storage/appservice.py @@ -343,15 +343,28 @@ class ApplicationServiceTransactionStore(SQLBaseStore): def __init__(self, hs): super(ApplicationServiceTransactionStore, self).__init__(hs) - def get_failing_appservices(self): - """Get a list of application services which are down. + def get_appservices_by_state(self, state): + """Get a list of application services based on their state. + Args: + state(ApplicationServiceState): The state to filter on. Returns: A Deferred which resolves to a list of ApplicationServices, which may be empty. """ pass + def set_appservice_state(self, service, state): + """Set the application service state. + + Args: + service(ApplicationService): The service whose state to set. + state(ApplicationServiceState): The connectivity state to apply. + Returns: + A Deferred which resolves to True if the state was set successfully. + """ + pass + def complete_appservice_txn(self, txn_id, service): """Completes an application service transaction. diff --git a/tests/appservice/test_scheduler.py b/tests/appservice/test_scheduler.py index 1e3eb9e1cc..ec8f77c54b 100644 --- a/tests/appservice/test_scheduler.py +++ b/tests/appservice/test_scheduler.py @@ -57,7 +57,8 @@ class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase): self.assertEquals(1, txn.complete.call_count) # 2 because it needs to get None to know there are no more txns self.assertEquals(2, self.store.get_oldest_txn.call_count) - self.assertEquals(1, self.callback.call_count) + self.callback.assert_called_once_with(self.recoverer) + self.assertEquals(self.recoverer.service, self.service) def test_recover_retry_txn(self): txn = Mock() @@ -91,7 +92,7 @@ class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase): self.clock.advance_time(16000) self.assertEquals(1, txn.send.call_count) # new mock reset call count self.assertEquals(1, txn.complete.call_count) - self.assertEquals(1, self.callback.call_count) + self.callback.assert_called_once_with(self.recoverer) class ApplicationServiceSchedulerEventGrouperTestCase(unittest.TestCase): -- cgit 1.5.1 From 0354659f9d8b60b9edc78b0b597bceb52b8c7b2b Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Fri, 6 Mar 2015 16:09:05 +0000 Subject: Finish synapse.appservice.scheduler implementation. With tests to assert behaviour. Not hooked up yet. Stub datastore methods not implemented yet. --- synapse/appservice/__init__.py | 39 +++++++++++++ synapse/appservice/scheduler.py | 63 ++++---------------- synapse/storage/appservice.py | 22 +++++++ tests/appservice/test_scheduler.py | 115 ++++++++++++++++++++++++++++++++++++- 4 files changed, 186 insertions(+), 53 deletions(-) (limited to 'tests') diff --git a/synapse/appservice/__init__.py b/synapse/appservice/__init__.py index cc6c381566..743a8278ad 100644 --- a/synapse/appservice/__init__.py +++ b/synapse/appservice/__init__.py @@ -25,6 +25,45 @@ class ApplicationServiceState(object): UP = "up" +class AppServiceTransaction(object): + """Represents an application service transaction.""" + + def __init__(self, service, id, events): + self.service = service + self.id = id + self.events = events + + def send(self, as_api): + """Sends this transaction using the provided AS API interface. + + Args: + as_api(ApplicationServiceApi): The API to use to send. + Returns: + A Deferred which resolves to True if the transaction was sent. + """ + return as_api.push_bulk( + service=self.service, + events=self.events, + txn_id=self.id + ) + + def complete(self, store): + """Completes this transaction as successful. + + Marks this transaction ID on the application service and removes the + transaction contents from the database. + + Args: + store: The database store to operate on. + Returns: + A Deferred which resolves to True if the transaction was completed. + """ + return store.complete_appservice_txn( + service=self.service, + txn_id=self.id + ) + + class ApplicationService(object): """Defines an application service. This definition is mostly what is provided to the /register AS API. diff --git a/synapse/appservice/scheduler.py b/synapse/appservice/scheduler.py index 2b3aa3b0ea..50ad3b8e83 100644 --- a/synapse/appservice/scheduler.py +++ b/synapse/appservice/scheduler.py @@ -88,45 +88,6 @@ class AppServiceScheduler(object): self.event_grouper.on_receive(service, event) -class AppServiceTransaction(object): - """Represents an application service transaction.""" - - def __init__(self, service, id, events): - self.service = service - self.id = id - self.events = events - - def send(self, as_api): - """Sends this transaction using the provided AS API interface. - - Args: - as_api(ApplicationServiceApi): The API to use to send. - Returns: - A Deferred which resolves to True if the transaction was sent. - """ - return as_api.push_bulk( - service=self.service, - events=self.events, - txn_id=self.id - ) - - def complete(self, store): - """Completes this transaction as successful. - - Marks this transaction ID on the application service and removes the - transaction contents from the database. - - Args: - store: The database store to operate on. - Returns: - A Deferred which resolves to True if the transaction was completed. - """ - return store.complete_appservice_txn( - service=self.service, - txn_id=self.id - ) - - class _EventGrouper(object): """Groups events for the same application service together. """ @@ -156,14 +117,18 @@ class _TransactionController(object): # keep track of how many recoverers there are self.recoverers = [] + @defer.inlineCallbacks def start_polling(self): groups = self.event_grouper.drain_groups() for service in groups: - txn_id = self._get_next_txn_id(service) - txn = AppServiceTransaction(service, txn_id, groups[service]) - self._store_txn(txn) - if self._is_service_up(service): - if txn.send(self.as_api): + txn = yield self.store.create_appservice_txn( + service=service, + events=groups[service] + ) + service_is_up = yield self._is_service_up(service) + if service_is_up: + sent = yield txn.send(self.as_api) + if sent: txn.complete(self.store) else: self._start_recoverer(service) @@ -207,14 +172,10 @@ class _TransactionController(object): logger.error("Failed to apply appservice state DOWN to service %s", service) + @defer.inlineCallbacks def _is_service_up(self, service): - pass - - def _get_next_txn_id(self, service): - pass # TODO work out the next txn_id for this service - - def _store_txn(self, txn): - pass + state = yield self.store.get_appservice_state(service) + defer.returnValue(state == ApplicationServiceState.UP) class _Recoverer(object): diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py index 214f6d99c5..6fde7dcc66 100644 --- a/synapse/storage/appservice.py +++ b/synapse/storage/appservice.py @@ -354,6 +354,16 @@ class ApplicationServiceTransactionStore(SQLBaseStore): """ pass + def get_appservice_state(self, service): + """Get the application service state. + + Args: + service(ApplicationService): The service whose state to set. + Returns: + A Deferred which resolves to ApplicationServiceState. + """ + pass + def set_appservice_state(self, service, state): """Set the application service state. @@ -365,6 +375,18 @@ class ApplicationServiceTransactionStore(SQLBaseStore): """ pass + def create_appservice_txn(self, service, events): + """Atomically creates a new transaction for this application service + with the given list of events. + + Args: + service(ApplicationService): The service who the transaction is for. + events(list): A list of events to put in the transaction. + Returns: + ApplicationServiceTransaction: A new transaction. + """ + pass + def complete_appservice_txn(self, txn_id, service): """Completes an application service transaction. diff --git a/tests/appservice/test_scheduler.py b/tests/appservice/test_scheduler.py index ec8f77c54b..a31755da67 100644 --- a/tests/appservice/test_scheduler.py +++ b/tests/appservice/test_scheduler.py @@ -12,9 +12,9 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +from synapse.appservice import ApplicationServiceState, AppServiceTransaction from synapse.appservice.scheduler import ( - AppServiceScheduler, AppServiceTransaction, _EventGrouper, - _TransactionController, _Recoverer + AppServiceScheduler, _EventGrouper, _TransactionController, _Recoverer ) from twisted.internet import defer from ..utils import MockClock @@ -22,6 +22,116 @@ from mock import Mock from tests import unittest +class ApplicationServiceSchedulerTransactionCtrlTestCase(unittest.TestCase): + + def setUp(self): + self.clock = MockClock() + self.store = Mock() + self.as_api = Mock() + self.event_grouper = Mock() + self.recoverer = Mock() + self.recoverer_fn = Mock(return_value=self.recoverer) + self.txnctrl = _TransactionController( + clock=self.clock, store=self.store, as_api=self.as_api, + event_grouper=self.event_grouper, recoverer_fn=self.recoverer_fn + ) + + def test_poll_single_group_service_up(self): + # Test: The AS is up and the txn is successfully sent. + service = Mock() + events = [Mock(), Mock()] + groups = {} + groups[service] = events + txn_id = "foobar" + txn = Mock(id=txn_id, service=service, events=events) + + # mock methods + self.event_grouper.drain_groups = Mock(return_value=groups) + self.store.get_appservice_state = Mock( + return_value=defer.succeed(ApplicationServiceState.UP) + ) + txn.send = Mock(return_value=defer.succeed(True)) + self.store.create_appservice_txn = Mock( + return_value=defer.succeed(txn) + ) + + # actual call + self.txnctrl.start_polling() + + self.store.create_appservice_txn.assert_called_once_with( + service=service, events=events # txn made and saved + ) + self.assertEquals(0, len(self.txnctrl.recoverers)) # no recoverer made + txn.complete.assert_called_once_with(self.store) # txn completed + + def test_poll_single_group_service_down(self): + # Test: The AS is down so it shouldn't push; Recoverers will do it. + # It should still make a transaction though. + service = Mock() + events = [Mock(), Mock()] + groups = {} + groups[service] = events + + self.event_grouper.drain_groups = Mock(return_value=groups) + txn = Mock(id="idhere", service=service, events=events) + self.store.get_appservice_state = Mock( + return_value=defer.succeed(ApplicationServiceState.DOWN) + ) + self.store.create_appservice_txn = Mock( + return_value=defer.succeed(txn) + ) + + # actual call + self.txnctrl.start_polling() + + self.store.create_appservice_txn.assert_called_once_with( + service=service, events=events # txn made and saved + ) + self.assertEquals(0, txn.send.call_count) # txn not sent though + self.assertEquals(0, txn.complete.call_count) # or completed + + def test_poll_single_group_service_up(self): + # Test: The AS is up and the txn is not sent. A Recoverer is made and + # started. + service = Mock() + events = [Mock(), Mock()] + groups = {} + groups[service] = events + txn_id = "foobar" + txn = Mock(id=txn_id, service=service, events=events) + + # mock methods + self.event_grouper.drain_groups = Mock(return_value=groups) + self.store.get_appservice_state = Mock( + return_value=defer.succeed(ApplicationServiceState.UP) + ) + self.store.set_appservice_state = Mock(return_value=defer.succeed(True)) + txn.send = Mock(return_value=defer.succeed(False)) # fails to send + self.store.create_appservice_txn = Mock( + return_value=defer.succeed(txn) + ) + + # actual call + self.txnctrl.start_polling() + + self.store.create_appservice_txn.assert_called_once_with( + service=service, events=events + ) + self.assertEquals(1, self.recoverer_fn.call_count) # recoverer made + self.assertEquals(1, self.recoverer.recover.call_count) # and invoked + self.assertEquals(1, len(self.txnctrl.recoverers)) # and stored + self.assertEquals(0, txn.complete.call_count) # txn not completed + self.store.set_appservice_state.assert_called_once_with( + service, ApplicationServiceState.DOWN # service marked as down + ) + + def test_poll_no_groups(self): + self.as_api.push_bulk = Mock() + self.event_grouper.drain_groups = Mock(return_value={}) + self.txnctrl.start_polling() + self.assertEquals(0, self.as_api.push_bulk.call_count) + + class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase): def setUp(self): @@ -94,6 +204,7 @@ class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase): self.assertEquals(1, txn.complete.call_count) self.callback.assert_called_once_with(self.recoverer) + class ApplicationServiceSchedulerEventGrouperTestCase(unittest.TestCase): def setUp(self): -- cgit 1.5.1 From 2602ddc379f9bede21cafc8c8f7f57dec44cf69d Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Fri, 6 Mar 2015 16:16:14 +0000 Subject: Apply clarity and docstrings --- synapse/appservice/scheduler.py | 2 +- synapse/storage/appservice.py | 14 +++++++++++++- tests/appservice/test_scheduler.py | 2 +- 3 files changed, 15 insertions(+), 3 deletions(-) (limited to 'tests') diff --git a/synapse/appservice/scheduler.py b/synapse/appservice/scheduler.py index 50ad3b8e83..514148c947 100644 --- a/synapse/appservice/scheduler.py +++ b/synapse/appservice/scheduler.py @@ -224,5 +224,5 @@ class _Recoverer(object): @defer.inlineCallbacks def _get_oldest_txn(self): - txn = yield self.store.get_oldest_txn(self.service) + txn = yield self.store.get_oldest_unsent_txn(self.service) defer.returnValue(txn) diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py index 6fde7dcc66..4447c8a2e1 100644 --- a/synapse/storage/appservice.py +++ b/synapse/storage/appservice.py @@ -383,7 +383,7 @@ class ApplicationServiceTransactionStore(SQLBaseStore): service(ApplicationService): The service who the transaction is for. events(list): A list of events to put in the transaction. Returns: - ApplicationServiceTransaction: A new transaction. + AppServiceTransaction: A new transaction. """ pass @@ -399,3 +399,15 @@ class ApplicationServiceTransactionStore(SQLBaseStore): successfully. """ pass + + def get_oldest_unsent_txn(self, service): + """Get the oldest transaction which has not been sent for this + service. + + Args: + service(ApplicationService): The app service to get the oldest txn. + Returns: + A Deferred which resolves to an AppServiceTransaction or + None. + """ + pass diff --git a/tests/appservice/test_scheduler.py b/tests/appservice/test_scheduler.py index a31755da67..f75a6f5d95 100644 --- a/tests/appservice/test_scheduler.py +++ b/tests/appservice/test_scheduler.py @@ -14,7 +14,7 @@ # limitations under the License. from synapse.appservice import ApplicationServiceState, AppServiceTransaction from synapse.appservice.scheduler import ( - AppServiceScheduler, _EventGrouper, _TransactionController, _Recoverer + _EventGrouper, _TransactionController, _Recoverer ) from twisted.internet import defer from ..utils import MockClock -- cgit 1.5.1 From 10766f1e93f884d8a71af43f565183a54786a3ca Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Fri, 6 Mar 2015 16:17:01 +0000 Subject: Update UTs --- tests/appservice/test_scheduler.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) (limited to 'tests') diff --git a/tests/appservice/test_scheduler.py b/tests/appservice/test_scheduler.py index f75a6f5d95..9532bf66b8 100644 --- a/tests/appservice/test_scheduler.py +++ b/tests/appservice/test_scheduler.py @@ -155,18 +155,18 @@ class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase): def take_txn(*args, **kwargs): return defer.succeed(txns.pop(0)) - self.store.get_oldest_txn = Mock(side_effect=take_txn) + self.store.get_oldest_unsent_txn = Mock(side_effect=take_txn) self.recoverer.recover() # shouldn't have called anything prior to waiting for exp backoff - self.assertEquals(0, self.store.get_oldest_txn.call_count) + self.assertEquals(0, self.store.get_oldest_unsent_txn.call_count) txn.send = Mock(return_value=True) # wait for exp backoff self.clock.advance_time(2000) self.assertEquals(1, txn.send.call_count) self.assertEquals(1, txn.complete.call_count) # 2 because it needs to get None to know there are no more txns - self.assertEquals(2, self.store.get_oldest_txn.call_count) + self.assertEquals(2, self.store.get_oldest_unsent_txn.call_count) self.callback.assert_called_once_with(self.recoverer) self.assertEquals(self.recoverer.service, self.service) @@ -180,10 +180,10 @@ class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase): return defer.succeed(txns.pop(0)) else: return defer.succeed(txn) - self.store.get_oldest_txn = Mock(side_effect=take_txn) + self.store.get_oldest_unsent_txn = Mock(side_effect=take_txn) self.recoverer.recover() - self.assertEquals(0, self.store.get_oldest_txn.call_count) + self.assertEquals(0, self.store.get_oldest_unsent_txn.call_count) txn.send = Mock(return_value=False) self.clock.advance_time(2000) self.assertEquals(1, txn.send.call_count) -- cgit 1.5.1 From 1c2dcf762a8fe28390e9a98a01577aaadca7f1c0 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Mon, 9 Mar 2015 13:10:31 +0000 Subject: Partially implement txn store methods with tests. --- synapse/storage/appservice.py | 61 ++++++++++---- tests/storage/test_appservice.py | 171 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 213 insertions(+), 19 deletions(-) (limited to 'tests') diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py index 0b272e82dd..37078f9ef0 100644 --- a/synapse/storage/appservice.py +++ b/synapse/storage/appservice.py @@ -13,13 +13,13 @@ # See the License for the specific language governing permissions and # limitations under the License. import logging -import simplejson from simplejson import JSONDecodeError +import simplejson as json from twisted.internet import defer from synapse.api.constants import Membership from synapse.api.errors import StoreError -from synapse.appservice import ApplicationService +from synapse.appservice import ApplicationService, ApplicationServiceState from synapse.storage.roommember import RoomsForUser from ._base import SQLBaseStore @@ -142,7 +142,7 @@ class ApplicationServiceStore(SQLBaseStore): txn.execute( "INSERT INTO application_services_regex(" "as_id, namespace, regex) values(?,?,?)", - (as_id, ns_int, simplejson.dumps(regex_obj)) + (as_id, ns_int, json.dumps(regex_obj)) ) return True @@ -277,12 +277,7 @@ class ApplicationServiceStore(SQLBaseStore): return rooms_for_user_matching_user_id - @defer.inlineCallbacks - def _populate_cache(self): - """Populates the ApplicationServiceCache from the database.""" - sql = ("SELECT * FROM application_services LEFT JOIN " - "application_services_regex ON application_services.id = " - "application_services_regex.as_id") + def _parse_services_dict(self, results): # SQL results in the form: # [ # { @@ -296,13 +291,12 @@ class ApplicationServiceStore(SQLBaseStore): # } # ] services = {} - results = yield self._execute_and_decode(sql) for res in results: as_token = res["token"] if as_token not in services: # add the service services[as_token] = { - "id": res["as_id"], + "id": res["id"], "url": res["url"], "token": as_token, "hs_token": res["hs_token"], @@ -320,16 +314,16 @@ class ApplicationServiceStore(SQLBaseStore): try: services[as_token]["namespaces"][ ApplicationService.NS_LIST[ns_int]].append( - simplejson.loads(res["regex"]) + json.loads(res["regex"]) ) except IndexError: logger.error("Bad namespace enum '%s'. %s", ns_int, res) except JSONDecodeError: logger.error("Bad regex object '%s'", res["regex"]) + service_list = [] for service in services.values(): - logger.info("Found application service: %s", service) - self.services_cache.append(ApplicationService( + service_list.append(ApplicationService( token=service["token"], url=service["url"], namespaces=service["namespaces"], @@ -337,6 +331,21 @@ class ApplicationServiceStore(SQLBaseStore): sender=service["sender"], id=service["id"] )) + return service_list + + @defer.inlineCallbacks + def _populate_cache(self): + """Populates the ApplicationServiceCache from the database.""" + sql = ("SELECT * FROM application_services LEFT JOIN " + "application_services_regex ON application_services.id = " + "application_services_regex.as_id") + + results = yield self._execute_and_decode(sql) + services = self._parse_services_dict(results) + + for service in services: + logger.info("Found application service: %s", service) + self.services_cache.append(service) class ApplicationServiceTransactionStore(SQLBaseStore): @@ -344,6 +353,7 @@ class ApplicationServiceTransactionStore(SQLBaseStore): def __init__(self, hs): super(ApplicationServiceTransactionStore, self).__init__(hs) + @defer.inlineCallbacks def get_appservices_by_state(self, state): """Get a list of application services based on their state. @@ -353,8 +363,16 @@ class ApplicationServiceTransactionStore(SQLBaseStore): A Deferred which resolves to a list of ApplicationServices, which may be empty. """ - pass + sql = ( + "SELECT r.*, a.* FROM application_services_state AS s LEFT JOIN " + "application_services AS a ON a.id=s.as_id LEFT JOIN " + "application_services_regex AS r ON r.as_id=a.id WHERE state = ?" + ) + results = yield self._execute_and_decode(sql, state) + # NB: This assumes this class is linked with ApplicationServiceStore + defer.returnValue(self._parse_services_dict(results)) + @defer.inlineCallbacks def get_appservice_state(self, service): """Get the application service state. @@ -363,7 +381,16 @@ class ApplicationServiceTransactionStore(SQLBaseStore): Returns: A Deferred which resolves to ApplicationServiceState. """ - pass + result = yield self._simple_select_one( + "application_services_state", + dict(as_id=service.id), + ["state"], + allow_none=True + ) + if result: + defer.returnValue(result.get("state")) + return + defer.returnValue(None) def set_appservice_state(self, service, state): """Set the application service state. @@ -372,7 +399,7 @@ class ApplicationServiceTransactionStore(SQLBaseStore): service(ApplicationService): The service whose state to set. state(ApplicationServiceState): The connectivity state to apply. Returns: - A Deferred which resolves to True if the state was set successfully. + A Deferred which resolves when the state was set successfully. """ return self._simple_upsert( "application_services_state", diff --git a/tests/storage/test_appservice.py b/tests/storage/test_appservice.py index ca5b92ec85..30c0b43d96 100644 --- a/tests/storage/test_appservice.py +++ b/tests/storage/test_appservice.py @@ -15,9 +15,11 @@ from tests import unittest from twisted.internet import defer -from synapse.appservice import ApplicationService +from synapse.appservice import ApplicationService, ApplicationServiceState from synapse.server import HomeServer -from synapse.storage.appservice import ApplicationServiceStore +from synapse.storage.appservice import ( + ApplicationServiceStore, ApplicationServiceTransactionStore +) from mock import Mock from tests.utils import SQLiteMemoryDbPool, MockClock @@ -114,3 +116,168 @@ class ApplicationServiceStoreTestCase(unittest.TestCase): def test_retrieval_of_all_services(self): services = yield self.store.get_app_services() self.assertEquals(len(services), 3) + + +class ApplicationServiceTransactionStoreTestCase(unittest.TestCase): + + @defer.inlineCallbacks + def setUp(self): + self.db_pool = SQLiteMemoryDbPool() + yield self.db_pool.prepare() + hs = HomeServer( + "test", db_pool=self.db_pool, clock=MockClock(), config=Mock() + ) + self.as_list = [ + { + "token": "token1", + "url": "https://matrix-as.org", + "id": 3 + }, + { + "token": "alpha_tok", + "url": "https://alpha.com", + "id": 5 + }, + { + "token": "beta_tok", + "url": "https://beta.com", + "id": 6 + }, + { + "token": "delta_tok", + "url": "https://delta.com", + "id": 7 + }, + ] + for s in self.as_list: + yield self._add_service(s["id"], s["url"], s["token"]) + self.store = TestTransactionStore(hs) + + def _add_service(self, as_id, url, token): + return self.db_pool.runQuery( + "INSERT INTO application_services(id, url, token) VALUES(?,?,?)", + (as_id, url, token) + ) + + def _set_state(self, id, state, txn=None): + return self.db_pool.runQuery( + "INSERT INTO application_services_state(as_id, state, last_txn) " + "VALUES(?,?,?)", + (id, state, txn) + ) + + @defer.inlineCallbacks + def test_get_appservice_state_none(self): + service = Mock(id=999) + state = yield self.store.get_appservice_state(service) + self.assertEquals(None, state) + + @defer.inlineCallbacks + def test_get_appservice_state_up(self): + yield self._set_state( + self.as_list[0]["id"], ApplicationServiceState.UP + ) + service = Mock(id=self.as_list[0]["id"]) + state = yield self.store.get_appservice_state(service) + self.assertEquals(ApplicationServiceState.UP, state) + + @defer.inlineCallbacks + def test_get_appservice_state_down(self): + yield self._set_state( + self.as_list[0]["id"], ApplicationServiceState.UP + ) + yield self._set_state( + self.as_list[1]["id"], ApplicationServiceState.DOWN + ) + yield self._set_state( + self.as_list[2]["id"], ApplicationServiceState.DOWN + ) + service = Mock(id=self.as_list[1]["id"]) + state = yield self.store.get_appservice_state(service) + self.assertEquals(ApplicationServiceState.DOWN, state) + + @defer.inlineCallbacks + def test_get_appservices_by_state_none(self): + services = yield self.store.get_appservices_by_state( + ApplicationServiceState.DOWN + ) + self.assertEquals(0, len(services)) + + @defer.inlineCallbacks + def test_set_appservices_state_down(self): + service = Mock(id=self.as_list[1]["id"]) + yield self.store.set_appservice_state( + service, + ApplicationServiceState.DOWN + ) + rows = yield self.db_pool.runQuery( + "SELECT as_id FROM application_services_state WHERE state=?", + (ApplicationServiceState.DOWN,) + ) + self.assertEquals(service.id, rows[0][0]) + + @defer.inlineCallbacks + def test_set_appservices_state_multiple_up(self): + service = Mock(id=self.as_list[1]["id"]) + yield self.store.set_appservice_state( + service, + ApplicationServiceState.UP + ) + yield self.store.set_appservice_state( + service, + ApplicationServiceState.DOWN + ) + yield self.store.set_appservice_state( + service, + ApplicationServiceState.UP + ) + rows = yield self.db_pool.runQuery( + "SELECT as_id FROM application_services_state WHERE state=?", + (ApplicationServiceState.UP,) + ) + self.assertEquals(service.id, rows[0][0]) + + @defer.inlineCallbacks + def test_get_appservices_by_state_single(self): + yield self._set_state( + self.as_list[0]["id"], ApplicationServiceState.DOWN + ) + yield self._set_state( + self.as_list[1]["id"], ApplicationServiceState.UP + ) + + services = yield self.store.get_appservices_by_state( + ApplicationServiceState.DOWN + ) + self.assertEquals(1, len(services)) + self.assertEquals(self.as_list[0]["id"], services[0].id) + + @defer.inlineCallbacks + def test_get_appservices_by_state_multiple(self): + yield self._set_state( + self.as_list[0]["id"], ApplicationServiceState.DOWN + ) + yield self._set_state( + self.as_list[1]["id"], ApplicationServiceState.UP + ) + yield self._set_state( + self.as_list[2]["id"], ApplicationServiceState.DOWN + ) + yield self._set_state( + self.as_list[3]["id"], ApplicationServiceState.UP + ) + + services = yield self.store.get_appservices_by_state( + ApplicationServiceState.DOWN + ) + self.assertEquals(2, len(services)) + self.assertEquals(self.as_list[2]["id"], services[0].id) + self.assertEquals(self.as_list[0]["id"], services[1].id) + + +# required for ApplicationServiceTransactionStoreTestCase tests +class TestTransactionStore(ApplicationServiceTransactionStore, + ApplicationServiceStore): + + def __init__(self, hs): + super(TestTransactionStore, self).__init__(hs) -- cgit 1.5.1 From 1ead1caa18bdbf708446f1faa3d6f3dd13e63c29 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Mon, 9 Mar 2015 13:54:20 +0000 Subject: Implement create_appservice_txn with tests. --- synapse/storage/appservice.py | 46 ++++++++++++++++++++++++--- tests/storage/test_appservice.py | 67 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 109 insertions(+), 4 deletions(-) (limited to 'tests') diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py index 37078f9ef0..1360a00eae 100644 --- a/synapse/storage/appservice.py +++ b/synapse/storage/appservice.py @@ -17,9 +17,10 @@ from simplejson import JSONDecodeError import simplejson as json from twisted.internet import defer +from syutil.jsonutil import encode_canonical_json from synapse.api.constants import Membership from synapse.api.errors import StoreError -from synapse.appservice import ApplicationService, ApplicationServiceState +from synapse.appservice import ApplicationService, AppServiceTransaction from synapse.storage.roommember import RoomsForUser from ._base import SQLBaseStore @@ -417,9 +418,46 @@ class ApplicationServiceTransactionStore(SQLBaseStore): Returns: AppServiceTransaction: A new transaction. """ - # TODO: work out txn id (highest txn id for this service += 1) - # TODO: Within same db transaction, Insert new txn into txn table - pass + return self.runInteraction( + "create_appservice_txn", + self._create_appservice_txn, + service, events + ) + + def _create_appservice_txn(self, txn, service, events): + # work out new txn id (highest txn id for this service += 1) + # The highest id may be the last one sent (in which case it is last_txn) + # or it may be the highest in the txns list (which are waiting to be/are + # being sent) + result = txn.execute( + "SELECT last_txn FROM application_services_state WHERE as_id=?", + (service.id,) + ) + last_txn_id = result.fetchone() + if last_txn_id is None: # no row exists + last_txn_id = 0 + else: + last_txn_id = int(last_txn_id[0]) # select 'last_txn' col + + result = txn.execute( + "SELECT MAX(txn_id) FROM application_services_txns WHERE as_id=?", + (service.id,) + ) + highest_txn_id = result.fetchone()[0] + if highest_txn_id is None: + highest_txn_id = 0 + + new_txn_id = max(highest_txn_id, last_txn_id) + 1 + + # Insert new txn into txn table + txn.execute( + "INSERT INTO application_services_txns(as_id, txn_id, content) " + "VALUES(?,?,?)", + (service.id, new_txn_id, encode_canonical_json(events)) + ) + return AppServiceTransaction( + service=service, id=new_txn_id, events=events + ) def complete_appservice_txn(self, txn_id, service): """Completes an application service transaction. diff --git a/tests/storage/test_appservice.py b/tests/storage/test_appservice.py index 30c0b43d96..7a8cdb5593 100644 --- a/tests/storage/test_appservice.py +++ b/tests/storage/test_appservice.py @@ -21,6 +21,7 @@ from synapse.storage.appservice import ( ApplicationServiceStore, ApplicationServiceTransactionStore ) +import json from mock import Mock from tests.utils import SQLiteMemoryDbPool, MockClock @@ -166,6 +167,20 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase): (id, state, txn) ) + def _insert_txn(self, as_id, txn_id, content): + return self.db_pool.runQuery( + "INSERT INTO application_services_txns(as_id, txn_id, content) " + "VALUES(?,?,?)", + (as_id, txn_id, json.dumps(content)) + ) + + def _set_last_txn(self, as_id, txn_id): + return self.db_pool.runQuery( + "INSERT INTO application_services_state(as_id, last_txn, state) " + "VALUES(?,?,?)", + (as_id, txn_id, ApplicationServiceState.UP) + ) + @defer.inlineCallbacks def test_get_appservice_state_none(self): service = Mock(id=999) @@ -237,6 +252,58 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase): ) self.assertEquals(service.id, rows[0][0]) + @defer.inlineCallbacks + def test_create_appservice_txn_first(self): + service = Mock(id=self.as_list[0]["id"]) + events = [{"type": "nothing"}, {"type": "here"}] + txn = yield self.store.create_appservice_txn(service, events) + self.assertEquals(txn.id, 1) + self.assertEquals(txn.events, events) + self.assertEquals(txn.service, service) + + @defer.inlineCallbacks + def test_create_appservice_txn_older_last_txn(self): + service = Mock(id=self.as_list[0]["id"]) + events = [{"type": "nothing"}, {"type": "here"}] + yield self._set_last_txn(service.id, 9643) # AS is falling behind + yield self._insert_txn(service.id, 9644, events) + yield self._insert_txn(service.id, 9645, events) + txn = yield self.store.create_appservice_txn(service, events) + self.assertEquals(txn.id, 9646) + self.assertEquals(txn.events, events) + self.assertEquals(txn.service, service) + + @defer.inlineCallbacks + def test_create_appservice_txn_up_to_date_last_txn(self): + service = Mock(id=self.as_list[0]["id"]) + events = [{"type": "nothing"}, {"type": "here"}] + yield self._set_last_txn(service.id, 9643) + txn = yield self.store.create_appservice_txn(service, events) + self.assertEquals(txn.id, 9644) + self.assertEquals(txn.events, events) + self.assertEquals(txn.service, service) + + @defer.inlineCallbacks + def test_create_appservice_txn_up_fuzzing(self): + service = Mock(id=self.as_list[0]["id"]) + events = [{"type": "nothing"}, {"type": "here"}] + yield self._set_last_txn(service.id, 9643) + + # dump in rows with higher IDs to make sure the queries aren't wrong. + yield self._set_last_txn(self.as_list[1]["id"], 119643) + yield self._set_last_txn(self.as_list[2]["id"], 9) + yield self._set_last_txn(self.as_list[3]["id"], 9643) + yield self._insert_txn(self.as_list[1]["id"], 119644, events) + yield self._insert_txn(self.as_list[1]["id"], 119645, events) + yield self._insert_txn(self.as_list[1]["id"], 119646, events) + yield self._insert_txn(self.as_list[2]["id"], 10, events) + yield self._insert_txn(self.as_list[3]["id"], 9643, events) + + txn = yield self.store.create_appservice_txn(service, events) + self.assertEquals(txn.id, 9644) + self.assertEquals(txn.events, events) + self.assertEquals(txn.service, service) + @defer.inlineCallbacks def test_get_appservices_by_state_single(self): yield self._set_state( -- cgit 1.5.1 From 0a60bbf4fac4262da3fee702ca46d2f019597ef1 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Mon, 9 Mar 2015 15:53:03 +0000 Subject: Finish appservice txn storage impl and tests. --- synapse/storage/appservice.py | 85 ++++++++++++++++++---- .../storage/schema/delta/15/appservice_txns.sql | 2 +- tests/storage/test_appservice.py | 68 +++++++++++++++++ 3 files changed, 139 insertions(+), 16 deletions(-) (limited to 'tests') diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py index 1360a00eae..d89b0cc8c9 100644 --- a/synapse/storage/appservice.py +++ b/synapse/storage/appservice.py @@ -429,15 +429,7 @@ class ApplicationServiceTransactionStore(SQLBaseStore): # The highest id may be the last one sent (in which case it is last_txn) # or it may be the highest in the txns list (which are waiting to be/are # being sent) - result = txn.execute( - "SELECT last_txn FROM application_services_state WHERE as_id=?", - (service.id,) - ) - last_txn_id = result.fetchone() - if last_txn_id is None: # no row exists - last_txn_id = 0 - else: - last_txn_id = int(last_txn_id[0]) # select 'last_txn' col + last_txn_id = self._get_last_txn(txn, service.id) result = txn.execute( "SELECT MAX(txn_id) FROM application_services_txns WHERE as_id=?", @@ -467,12 +459,43 @@ class ApplicationServiceTransactionStore(SQLBaseStore): service(ApplicationService): The application service which was sent this transaction. Returns: - A Deferred which resolves to True if this transaction was completed + A Deferred which resolves if this transaction was stored successfully. """ - # TODO: Set current txn_id for AS to 'txn_id' - # TODO: Delete txn contents - pass + return self.runInteraction( + "complete_appservice_txn", + self._complete_appservice_txn, + txn_id, service + ) + + def _complete_appservice_txn(self, txn, txn_id, service): + txn_id = int(txn_id) + + # Debugging query: Make sure the txn being completed is EXACTLY +1 from + # what was there before. If it isn't, we've got problems (e.g. the AS + # has probably missed some events), so whine loudly but still continue, + # since it shouldn't fail completion of the transaction. + last_txn_id = self._get_last_txn(txn, service.id) + if (last_txn_id + 1) != txn_id: + logger.error( + "appservice: Completing a transaction which has an ID > 1 from " + "the last ID sent to this AS. We've either dropped events or " + "sent it to the AS out of order. FIX ME. last_txn=%s " + "completing_txn=%s service_id=%s", last_txn_id, txn_id, + service.id + ) + + # Set current txn_id for AS to 'txn_id' + self._simple_upsert_txn( + txn, "application_services_state", dict(as_id=service.id), + dict(last_txn=txn_id) + ) + + # Delete txn contents + self._simple_delete_txn( + txn, "application_services_txns", + dict(txn_id=txn_id, as_id=service.id) + ) def get_oldest_unsent_txn(self, service): """Get the oldest transaction which has not been sent for this @@ -484,6 +507,38 @@ class ApplicationServiceTransactionStore(SQLBaseStore): A Deferred which resolves to an AppServiceTransaction or None. """ - # TODO: Monotonically increasing txn ids, so just select the smallest + return self.runInteraction( + "get_oldest_unsent_appservice_txn", + self._get_oldest_unsent_txn, + service + ) + + def _get_oldest_unsent_txn(self, txn, service): + # Monotonically increasing txn ids, so just select the smallest # one in the txns table (we delete them when they are sent) - pass + result = txn.execute( + "SELECT *,MIN(txn_id) FROM application_services_txns WHERE as_id=?", + (service.id,) + ) + entry = self.cursor_to_dict(result)[0] + + if not entry or entry["txn_id"] is None: + # the min(txn_id) part will force a row, so entry may not be None + return None + + return AppServiceTransaction( + service=service, id=entry["txn_id"], events=json.loads( + entry["content"] + ) + ) + + def _get_last_txn(self, txn, service_id): + result = txn.execute( + "SELECT last_txn FROM application_services_state WHERE as_id=?", + (service_id,) + ) + last_txn_id = result.fetchone() + if last_txn_id is None: # no row exists + return 0 + else: + return int(last_txn_id[0]) # select 'last_txn' col diff --git a/synapse/storage/schema/delta/15/appservice_txns.sql b/synapse/storage/schema/delta/15/appservice_txns.sql index 11f0c799aa..ff15aa019e 100644 --- a/synapse/storage/schema/delta/15/appservice_txns.sql +++ b/synapse/storage/schema/delta/15/appservice_txns.sql @@ -15,7 +15,7 @@ CREATE TABLE IF NOT EXISTS application_services_state( as_id INTEGER PRIMARY KEY, - state TEXT NOT NULL, + state TEXT, last_txn TEXT, FOREIGN KEY(as_id) REFERENCES application_services(id) ); diff --git a/tests/storage/test_appservice.py b/tests/storage/test_appservice.py index 7a8cdb5593..d1809c7f3b 100644 --- a/tests/storage/test_appservice.py +++ b/tests/storage/test_appservice.py @@ -304,6 +304,74 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase): self.assertEquals(txn.events, events) self.assertEquals(txn.service, service) + @defer.inlineCallbacks + def test_complete_appservice_txn_first_txn(self): + service = Mock(id=self.as_list[0]["id"]) + events = [{"foo": "bar"}] + txn_id = 1 + + yield self._insert_txn(service.id, txn_id, events) + yield self.store.complete_appservice_txn(txn_id=txn_id, service=service) + + res = yield self.db_pool.runQuery( + "SELECT last_txn FROM application_services_state WHERE as_id=?", + (service.id,) + ) + self.assertEquals(1, len(res)) + self.assertEquals(str(txn_id), res[0][0]) + + res = yield self.db_pool.runQuery( + "SELECT * FROM application_services_txns WHERE txn_id=?", + (txn_id,) + ) + self.assertEquals(0, len(res)) + + @defer.inlineCallbacks + def test_complete_appservice_txn_existing_in_state_table(self): + service = Mock(id=self.as_list[0]["id"]) + events = [{"foo": "bar"}] + txn_id = 5 + yield self._set_last_txn(service.id, 4) + yield self._insert_txn(service.id, txn_id, events) + yield self.store.complete_appservice_txn(txn_id=txn_id, service=service) + + res = yield self.db_pool.runQuery( + "SELECT last_txn, state FROM application_services_state WHERE " + "as_id=?", + (service.id,) + ) + self.assertEquals(1, len(res)) + self.assertEquals(str(txn_id), res[0][0]) + self.assertEquals(ApplicationServiceState.UP, res[0][1]) + + res = yield self.db_pool.runQuery( + "SELECT * FROM application_services_txns WHERE txn_id=?", + (txn_id,) + ) + self.assertEquals(0, len(res)) + + @defer.inlineCallbacks + def test_get_oldest_unsent_txn_none(self): + service = Mock(id=self.as_list[0]["id"]) + + txn = yield self.store.get_oldest_unsent_txn(service) + self.assertEquals(None, txn) + + @defer.inlineCallbacks + def test_get_oldest_unsent_txn(self): + service = Mock(id=self.as_list[0]["id"]) + events = [{"type": "nothing"}, {"type": "here"}] + + yield self._insert_txn(self.as_list[1]["id"], 9, {"badger": "mushroom"}) + yield self._insert_txn(service.id, 10, events) + yield self._insert_txn(service.id, 11, [{"foo":"bar"}]) + yield self._insert_txn(service.id, 12, [{"argh":"bargh"}]) + + txn = yield self.store.get_oldest_unsent_txn(service) + self.assertEquals(service, txn.service) + self.assertEquals(10, txn.id) + self.assertEquals(events, txn.events) + @defer.inlineCallbacks def test_get_appservices_by_state_single(self): yield self._set_state( -- cgit 1.5.1 From 21fd84dcb8645a555cc35adb8b2a5a68536b8087 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Mon, 9 Mar 2015 17:01:19 +0000 Subject: Use seconds; start gluing in the AS scheduler into the AS handler. --- synapse/appservice/scheduler.py | 4 ++-- synapse/handlers/__init__.py | 8 +++++++- synapse/handlers/appservice.py | 17 ++++++++++++++--- synapse/storage/__init__.py | 7 +++++-- tests/appservice/test_scheduler.py | 10 +++++----- tests/handlers/test_appservice.py | 7 +++++-- 6 files changed, 38 insertions(+), 15 deletions(-) (limited to 'tests') diff --git a/synapse/appservice/scheduler.py b/synapse/appservice/scheduler.py index ee5978da6e..068d4bd087 100644 --- a/synapse/appservice/scheduler.py +++ b/synapse/appservice/scheduler.py @@ -132,7 +132,7 @@ class _TransactionController(object): txn.complete(self.store) else: self._start_recoverer(service) - self.clock.call_later(1000, self.start_polling) + self.clock.call_later(1, self.start_polling) @defer.inlineCallbacks def on_recovered(self, recoverer): @@ -202,7 +202,7 @@ class _Recoverer(object): self.backoff_counter = 1 def recover(self): - self.clock.call_later(1000 * (2 ** self.backoff_counter), self.retry) + self.clock.call_later((2 ** self.backoff_counter), self.retry) @defer.inlineCallbacks def retry(self): diff --git a/synapse/handlers/__init__.py b/synapse/handlers/__init__.py index 8d345bf936..0c51d615ec 100644 --- a/synapse/handlers/__init__.py +++ b/synapse/handlers/__init__.py @@ -13,6 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +from synapse.appservice.scheduler import AppServiceScheduler from synapse.appservice.api import ApplicationServiceApi from .register import RegistrationHandler from .room import ( @@ -54,7 +55,12 @@ class Handlers(object): self.directory_handler = DirectoryHandler(hs) self.typing_notification_handler = TypingNotificationHandler(hs) self.admin_handler = AdminHandler(hs) + asapi = ApplicationServiceApi(hs) self.appservice_handler = ApplicationServicesHandler( - hs, ApplicationServiceApi(hs) + hs, asapi, AppServiceScheduler( + clock=hs.get_clock(), + store=hs.get_datastore(), + as_api=asapi + ) ) self.sync_handler = SyncHandler(hs) diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py index 2c488a46f6..f3cd458e6b 100644 --- a/synapse/handlers/appservice.py +++ b/synapse/handlers/appservice.py @@ -26,15 +26,22 @@ import logging logger = logging.getLogger(__name__) +def log_failure(failure): + logger.error("Application Services Failure: %s", failure.value) + logger.error(failure.getTraceback()) + + # NB: Purposefully not inheriting BaseHandler since that contains way too much # setup code which this handler does not need or use. This makes testing a lot # easier. class ApplicationServicesHandler(object): - def __init__(self, hs, appservice_api): + def __init__(self, hs, appservice_api, appservice_scheduler): self.store = hs.get_datastore() self.hs = hs self.appservice_api = appservice_api + self.scheduler = appservice_scheduler + self.started_scheduler = False @defer.inlineCallbacks def register(self, app_service): @@ -90,9 +97,13 @@ class ApplicationServicesHandler(object): if event.type == EventTypes.Member: yield self._check_user_exists(event.state_key) - # Fork off pushes to these services - XXX First cut, best effort + if not self.started_scheduler: + self.scheduler.start().addErrback(log_failure) + self.started_scheduler = True + + # Fork off pushes to these services for service in services: - self.appservice_api.push(service, event) + self.scheduler.submit_event_for_as(service, event) @defer.inlineCallbacks def query_user_exists(self, user_id): diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index dfce5224a9..6c159b52a0 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -18,7 +18,9 @@ from twisted.internet import defer from synapse.util.logutils import log_function from synapse.api.constants import EventTypes -from .appservice import ApplicationServiceStore +from .appservice import ( + ApplicationServiceStore, ApplicationServiceTransactionStore +) from .directory import DirectoryStore from .feedback import FeedbackStore from .presence import PresenceStore @@ -79,7 +81,8 @@ class DataStore(RoomMemberStore, RoomStore, RejectionsStore, FilteringStore, PusherStore, - PushRuleStore + PushRuleStore, + ApplicationServiceTransactionStore ): def __init__(self, hs): diff --git a/tests/appservice/test_scheduler.py b/tests/appservice/test_scheduler.py index 9532bf66b8..e18e879319 100644 --- a/tests/appservice/test_scheduler.py +++ b/tests/appservice/test_scheduler.py @@ -162,7 +162,7 @@ class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase): self.assertEquals(0, self.store.get_oldest_unsent_txn.call_count) txn.send = Mock(return_value=True) # wait for exp backoff - self.clock.advance_time(2000) + self.clock.advance_time(2) self.assertEquals(1, txn.send.call_count) self.assertEquals(1, txn.complete.call_count) # 2 because it needs to get None to know there are no more txns @@ -185,21 +185,21 @@ class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase): self.recoverer.recover() self.assertEquals(0, self.store.get_oldest_unsent_txn.call_count) txn.send = Mock(return_value=False) - self.clock.advance_time(2000) + self.clock.advance_time(2) self.assertEquals(1, txn.send.call_count) self.assertEquals(0, txn.complete.call_count) self.assertEquals(0, self.callback.call_count) - self.clock.advance_time(4000) + self.clock.advance_time(4) self.assertEquals(2, txn.send.call_count) self.assertEquals(0, txn.complete.call_count) self.assertEquals(0, self.callback.call_count) - self.clock.advance_time(8000) + self.clock.advance_time(8) self.assertEquals(3, txn.send.call_count) self.assertEquals(0, txn.complete.call_count) self.assertEquals(0, self.callback.call_count) txn.send = Mock(return_value=True) # successfully send the txn pop_txn = True # returns the txn the first time, then no more. - self.clock.advance_time(16000) + self.clock.advance_time(16) self.assertEquals(1, txn.send.call_count) # new mock reset call count self.assertEquals(1, txn.complete.call_count) self.callback.assert_called_once_with(self.recoverer) diff --git a/tests/handlers/test_appservice.py b/tests/handlers/test_appservice.py index a2c541317c..06cb1dd4cf 100644 --- a/tests/handlers/test_appservice.py +++ b/tests/handlers/test_appservice.py @@ -27,10 +27,11 @@ class AppServiceHandlerTestCase(unittest.TestCase): def setUp(self): self.mock_store = Mock() self.mock_as_api = Mock() + self.mock_scheduler = Mock() hs = Mock() hs.get_datastore = Mock(return_value=self.mock_store) self.handler = ApplicationServicesHandler( - hs, self.mock_as_api + hs, self.mock_as_api, self.mock_scheduler ) @defer.inlineCallbacks @@ -52,7 +53,9 @@ class AppServiceHandlerTestCase(unittest.TestCase): ) self.mock_as_api.push = Mock() yield self.handler.notify_interested_services(event) - self.mock_as_api.push.assert_called_once_with(interested_service, event) + self.mock_scheduler.submit_event_for_as.assert_called_once_with( + interested_service, event + ) @defer.inlineCallbacks def test_query_room_alias_exists(self): -- cgit 1.5.1 From b98cd03193476dea5f8b47e79d4122bb18449ae2 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Mon, 9 Mar 2015 17:25:20 +0000 Subject: Use event IDs instead of dumping event content in the txns table. --- synapse/storage/appservice.py | 14 +++++----- .../storage/schema/delta/15/appservice_txns.sql | 2 +- tests/storage/test_appservice.py | 30 ++++++++++++---------- 3 files changed, 26 insertions(+), 20 deletions(-) (limited to 'tests') diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py index d89b0cc8c9..c3c0a0bd43 100644 --- a/synapse/storage/appservice.py +++ b/synapse/storage/appservice.py @@ -442,10 +442,11 @@ class ApplicationServiceTransactionStore(SQLBaseStore): new_txn_id = max(highest_txn_id, last_txn_id) + 1 # Insert new txn into txn table + event_ids = [e.event_id for e in events] txn.execute( - "INSERT INTO application_services_txns(as_id, txn_id, content) " + "INSERT INTO application_services_txns(as_id, txn_id, event_ids) " "VALUES(?,?,?)", - (service.id, new_txn_id, encode_canonical_json(events)) + (service.id, new_txn_id, json.dumps(event_ids)) ) return AppServiceTransaction( service=service, id=new_txn_id, events=events @@ -491,7 +492,7 @@ class ApplicationServiceTransactionStore(SQLBaseStore): dict(last_txn=txn_id) ) - # Delete txn contents + # Delete txn self._simple_delete_txn( txn, "application_services_txns", dict(txn_id=txn_id, as_id=service.id) @@ -526,10 +527,11 @@ class ApplicationServiceTransactionStore(SQLBaseStore): # the min(txn_id) part will force a row, so entry may not be None return None + event_ids = json.loads(entry["event_ids"]) + events = self._get_events_txn(event_ids) + return AppServiceTransaction( - service=service, id=entry["txn_id"], events=json.loads( - entry["content"] - ) + service=service, id=entry["txn_id"], events=events ) def _get_last_txn(self, txn, service_id): diff --git a/synapse/storage/schema/delta/15/appservice_txns.sql b/synapse/storage/schema/delta/15/appservice_txns.sql index ff15aa019e..13bbb2de2e 100644 --- a/synapse/storage/schema/delta/15/appservice_txns.sql +++ b/synapse/storage/schema/delta/15/appservice_txns.sql @@ -23,7 +23,7 @@ CREATE TABLE IF NOT EXISTS application_services_state( CREATE TABLE IF NOT EXISTS application_services_txns( as_id INTEGER NOT NULL, txn_id INTEGER NOT NULL, - content TEXT NOT NULL, + event_ids TEXT NOT NULL, UNIQUE(as_id, txn_id) ON CONFLICT ROLLBACK ); diff --git a/tests/storage/test_appservice.py b/tests/storage/test_appservice.py index d1809c7f3b..e79599f7fb 100644 --- a/tests/storage/test_appservice.py +++ b/tests/storage/test_appservice.py @@ -167,11 +167,11 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase): (id, state, txn) ) - def _insert_txn(self, as_id, txn_id, content): + def _insert_txn(self, as_id, txn_id, events): return self.db_pool.runQuery( - "INSERT INTO application_services_txns(as_id, txn_id, content) " + "INSERT INTO application_services_txns(as_id, txn_id, event_ids) " "VALUES(?,?,?)", - (as_id, txn_id, json.dumps(content)) + (as_id, txn_id, json.dumps([e.event_id for e in events])) ) def _set_last_txn(self, as_id, txn_id): @@ -255,7 +255,7 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase): @defer.inlineCallbacks def test_create_appservice_txn_first(self): service = Mock(id=self.as_list[0]["id"]) - events = [{"type": "nothing"}, {"type": "here"}] + events = [Mock(event_id="e1"), Mock(event_id="e2")] txn = yield self.store.create_appservice_txn(service, events) self.assertEquals(txn.id, 1) self.assertEquals(txn.events, events) @@ -264,7 +264,7 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase): @defer.inlineCallbacks def test_create_appservice_txn_older_last_txn(self): service = Mock(id=self.as_list[0]["id"]) - events = [{"type": "nothing"}, {"type": "here"}] + events = [Mock(event_id="e1"), Mock(event_id="e2")] yield self._set_last_txn(service.id, 9643) # AS is falling behind yield self._insert_txn(service.id, 9644, events) yield self._insert_txn(service.id, 9645, events) @@ -276,7 +276,7 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase): @defer.inlineCallbacks def test_create_appservice_txn_up_to_date_last_txn(self): service = Mock(id=self.as_list[0]["id"]) - events = [{"type": "nothing"}, {"type": "here"}] + events = [Mock(event_id="e1"), Mock(event_id="e2")] yield self._set_last_txn(service.id, 9643) txn = yield self.store.create_appservice_txn(service, events) self.assertEquals(txn.id, 9644) @@ -286,7 +286,7 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase): @defer.inlineCallbacks def test_create_appservice_txn_up_fuzzing(self): service = Mock(id=self.as_list[0]["id"]) - events = [{"type": "nothing"}, {"type": "here"}] + events = [Mock(event_id="e1"), Mock(event_id="e2")] yield self._set_last_txn(service.id, 9643) # dump in rows with higher IDs to make sure the queries aren't wrong. @@ -307,7 +307,7 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase): @defer.inlineCallbacks def test_complete_appservice_txn_first_txn(self): service = Mock(id=self.as_list[0]["id"]) - events = [{"foo": "bar"}] + events = [Mock(event_id="e1"), Mock(event_id="e2")] txn_id = 1 yield self._insert_txn(service.id, txn_id, events) @@ -329,7 +329,7 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase): @defer.inlineCallbacks def test_complete_appservice_txn_existing_in_state_table(self): service = Mock(id=self.as_list[0]["id"]) - events = [{"foo": "bar"}] + events = [Mock(event_id="e1"), Mock(event_id="e2")] txn_id = 5 yield self._set_last_txn(service.id, 4) yield self._insert_txn(service.id, txn_id, events) @@ -360,12 +360,16 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase): @defer.inlineCallbacks def test_get_oldest_unsent_txn(self): service = Mock(id=self.as_list[0]["id"]) - events = [{"type": "nothing"}, {"type": "here"}] + events = [Mock(event_id="e1"), Mock(event_id="e2")] + other_events = [Mock(event_id="e5"), Mock(event_id="e6")] - yield self._insert_txn(self.as_list[1]["id"], 9, {"badger": "mushroom"}) + # we aren't testing store._base stuff here, so mock this out + self.store._get_events_txn = Mock(return_value=events) + + yield self._insert_txn(self.as_list[1]["id"], 9, other_events) yield self._insert_txn(service.id, 10, events) - yield self._insert_txn(service.id, 11, [{"foo":"bar"}]) - yield self._insert_txn(service.id, 12, [{"argh":"bargh"}]) + yield self._insert_txn(service.id, 11, other_events) + yield self._insert_txn(service.id, 12, other_events) txn = yield self.store.get_oldest_unsent_txn(service) self.assertEquals(service, txn.service) -- cgit 1.5.1 From 835e01fc7047e34a813936544027596627a112df Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Mon, 16 Mar 2015 10:16:59 +0000 Subject: Minor PR comment tweaks. --- synapse/appservice/scheduler.py | 4 ++-- synapse/handlers/appservice.py | 10 ++++++++-- synapse/storage/__init__.py | 2 +- synapse/storage/appservice.py | 6 +++--- tests/appservice/test_scheduler.py | 10 +++++----- 5 files changed, 19 insertions(+), 13 deletions(-) (limited to 'tests') diff --git a/synapse/appservice/scheduler.py b/synapse/appservice/scheduler.py index add1e3879c..8a3a6a880f 100644 --- a/synapse/appservice/scheduler.py +++ b/synapse/appservice/scheduler.py @@ -86,7 +86,7 @@ class AppServiceScheduler(object): self.txn_ctrl.start_polling() def submit_event_for_as(self, service, event): - self.event_grouper.on_receive(service, event) + self.event_grouper.enqueue(service, event) class _EventGrouper(object): @@ -96,7 +96,7 @@ class _EventGrouper(object): def __init__(self): self.groups = {} # dict of {service: [events]} - def on_receive(self, service, event): + def enqueue(self, service, event): if service not in self.groups: self.groups[service] = [] self.groups[service].append(event) diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py index f3cd458e6b..a24f7f5587 100644 --- a/synapse/handlers/appservice.py +++ b/synapse/handlers/appservice.py @@ -27,8 +27,14 @@ logger = logging.getLogger(__name__) def log_failure(failure): - logger.error("Application Services Failure: %s", failure.value) - logger.error(failure.getTraceback()) + logger.error( + "Application Services Failure", + exc_info=( + failure.type, + failure.value, + failure.getTracebackObject() + ) + ) # NB: Purposefully not inheriting BaseHandler since that contains way too much diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index efef859214..e752b035e6 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -82,7 +82,7 @@ class DataStore(RoomMemberStore, RoomStore, FilteringStore, PusherStore, PushRuleStore, - ApplicationServiceTransactionStore + ApplicationServiceTransactionStore, ): def __init__(self, hs): diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py index 670e1d56af..e928812bc9 100644 --- a/synapse/storage/appservice.py +++ b/synapse/storage/appservice.py @@ -365,9 +365,9 @@ class ApplicationServiceTransactionStore(SQLBaseStore): may be empty. """ sql = ( - "SELECT r.*, a.* FROM application_services_state AS s LEFT JOIN " - "application_services AS a ON a.id=s.as_id LEFT JOIN " - "application_services_regex AS r ON r.as_id=a.id WHERE state = ?" + "SELECT r.*, a.* FROM application_services_state AS s LEFT JOIN" + " application_services AS a ON a.id=s.as_id LEFT JOIN" + " application_services_regex AS r ON r.as_id=a.id WHERE state = ?" ) results = yield self._execute_and_decode( "get_appservices_by_state", sql, state diff --git a/tests/appservice/test_scheduler.py b/tests/appservice/test_scheduler.py index e18e879319..4534d05b93 100644 --- a/tests/appservice/test_scheduler.py +++ b/tests/appservice/test_scheduler.py @@ -213,7 +213,7 @@ class ApplicationServiceSchedulerEventGrouperTestCase(unittest.TestCase): def test_drain_single_event(self): service = Mock() event = Mock() - self.grouper.on_receive(service, event) + self.grouper.enqueue(service, event) groups = self.grouper.drain_groups() self.assertTrue(service in groups) self.assertEquals([event], groups[service]) @@ -225,7 +225,7 @@ class ApplicationServiceSchedulerEventGrouperTestCase(unittest.TestCase): service = Mock() events = [Mock(), Mock(), Mock()] for e in events: - self.grouper.on_receive(service, e) + self.grouper.enqueue(service, e) groups = self.grouper.drain_groups() self.assertTrue(service in groups) self.assertEquals(events, groups[service]) @@ -243,11 +243,11 @@ class ApplicationServiceSchedulerEventGrouperTestCase(unittest.TestCase): services[2]: events_c } for e in events_b: - self.grouper.on_receive(services[1], e) + self.grouper.enqueue(services[1], e) for e in events_c: - self.grouper.on_receive(services[2], e) + self.grouper.enqueue(services[2], e) for e in events_a: - self.grouper.on_receive(services[0], e) + self.grouper.enqueue(services[0], e) groups = self.grouper.drain_groups() for service in services: -- cgit 1.5.1 From 6279285b2ad59cf003b2e8d73d30dc706e1f3e4a Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Mon, 16 Mar 2015 13:15:40 +0000 Subject: Replace EventGrouper for ServiceQueuer to move to push-based txns. Fix tests and add stub tests for ServiceQueuer. --- synapse/appservice/scheduler.py | 61 +++++++++++----------- tests/appservice/test_scheduler.py | 100 ++++++++++--------------------------- 2 files changed, 60 insertions(+), 101 deletions(-) (limited to 'tests') diff --git a/synapse/appservice/scheduler.py b/synapse/appservice/scheduler.py index 59a870e271..54c42d1b94 100644 --- a/synapse/appservice/scheduler.py +++ b/synapse/appservice/scheduler.py @@ -16,11 +16,11 @@ This module controls the reliability for application service transactions. The nominal flow through this module looks like: - _________ ----ASa[e]-->| Event | -----ASb[e]->| Grouper |<-poll 1/s--+ ---ASa[e]--->|_________| | ASa[e,e] ASb[e] - V + __________ +1---ASa[e]-->| Service |--> Queue ASa[f] +2----ASb[e]->| Queuer | +3--ASa[f]--->|__________|-----------+ ASa[e], ASb[e] + V -````````- +------------+ |````````|<--StoreTxn-|Transaction | |Database| | Controller |---> SEND TO AS @@ -66,14 +66,14 @@ class AppServiceScheduler(object): self.clock = clock self.store = store self.as_api = as_api - self.event_grouper = _EventGrouper() def create_recoverer(service, callback): return _Recoverer(clock, store, as_api, service, callback) self.txn_ctrl = _TransactionController( - clock, store, as_api, self.event_grouper, create_recoverer + clock, store, as_api, create_recoverer ) + self.queuer = _ServiceQueuer(self.txn_ctrl) @defer.inlineCallbacks def start(self): @@ -86,17 +86,26 @@ class AppServiceScheduler(object): self.txn_ctrl.start_polling() def submit_event_for_as(self, service, event): - self.event_grouper.enqueue(service, event) + self.queuer.enqueue(service, event) -class _EventGrouper(object): - """Groups events for the same application service together. +class _ServiceQueuer(object): + """Queues events for the same application service together, sending + transactions as soon as possible. Once a transaction is sent successfully, + this schedules any other events in the queue to run. """ - def __init__(self): + def __init__(self, txn_ctrl): self.groups = {} # dict of {service: [events]} + self.txn_ctrl = txn_ctrl def enqueue(self, service, event): + # if nothing in queue for this service, send event immediately and add + # callbacks. + self.txn_ctrl.send(service, [event]) + + # else add to queue for this service + if service not in self.groups: self.groups[service] = [] self.groups[service].append(event) @@ -109,34 +118,30 @@ class _EventGrouper(object): class _TransactionController(object): - def __init__(self, clock, store, as_api, event_grouper, recoverer_fn): + def __init__(self, clock, store, as_api, recoverer_fn): self.clock = clock self.store = store self.as_api = as_api - self.event_grouper = event_grouper self.recoverer_fn = recoverer_fn # keep track of how many recoverers there are self.recoverers = [] @defer.inlineCallbacks - def start_polling(self): + def send(self, service, events): try: - groups = self.event_grouper.drain_groups() - for service in groups: - txn = yield self.store.create_appservice_txn( - service=service, - events=groups[service] - ) - service_is_up = yield self._is_service_up(service) - if service_is_up: - sent = yield txn.send(self.as_api) - if sent: - txn.complete(self.store) - else: - self._start_recoverer(service) + txn = yield self.store.create_appservice_txn( + service=service, + events=events + ) + service_is_up = yield self._is_service_up(service) + if service_is_up: + sent = yield txn.send(self.as_api) + if sent: + txn.complete(self.store) + else: + self._start_recoverer(service) except Exception as e: logger.exception(e) - self.clock.call_later(1, self.start_polling) @defer.inlineCallbacks def on_recovered(self, recoverer): diff --git a/tests/appservice/test_scheduler.py b/tests/appservice/test_scheduler.py index 4534d05b93..38d792eb02 100644 --- a/tests/appservice/test_scheduler.py +++ b/tests/appservice/test_scheduler.py @@ -14,7 +14,7 @@ # limitations under the License. from synapse.appservice import ApplicationServiceState, AppServiceTransaction from synapse.appservice.scheduler import ( - _EventGrouper, _TransactionController, _Recoverer + _ServiceQueuer, _TransactionController, _Recoverer ) from twisted.internet import defer from ..utils import MockClock @@ -28,25 +28,21 @@ class ApplicationServiceSchedulerTransactionCtrlTestCase(unittest.TestCase): self.clock = MockClock() self.store = Mock() self.as_api = Mock() - self.event_grouper = Mock() self.recoverer = Mock() self.recoverer_fn = Mock(return_value=self.recoverer) self.txnctrl = _TransactionController( clock=self.clock, store=self.store, as_api=self.as_api, - event_grouper=self.event_grouper, recoverer_fn=self.recoverer_fn + recoverer_fn=self.recoverer_fn ) - def test_poll_single_group_service_up(self): + def test_single_service_up_txn_sent(self): # Test: The AS is up and the txn is successfully sent. service = Mock() events = [Mock(), Mock()] - groups = {} - groups[service] = events txn_id = "foobar" txn = Mock(id=txn_id, service=service, events=events) # mock methods - self.event_grouper.drain_groups = Mock(return_value=groups) self.store.get_appservice_state = Mock( return_value=defer.succeed(ApplicationServiceState.UP) ) @@ -56,7 +52,7 @@ class ApplicationServiceSchedulerTransactionCtrlTestCase(unittest.TestCase): ) # actual call - self.txnctrl.start_polling() + self.txnctrl.send(service, events) self.store.create_appservice_txn.assert_called_once_with( service=service, events=events # txn made and saved @@ -64,15 +60,12 @@ class ApplicationServiceSchedulerTransactionCtrlTestCase(unittest.TestCase): self.assertEquals(0, len(self.txnctrl.recoverers)) # no recoverer made txn.complete.assert_called_once_with(self.store) # txn completed - def test_poll_single_group_service_down(self): + def test_single_service_down(self): # Test: The AS is down so it shouldn't push; Recoverers will do it. # It should still make a transaction though. service = Mock() events = [Mock(), Mock()] - groups = {} - groups[service] = events - self.event_grouper.drain_groups = Mock(return_value=groups) txn = Mock(id="idhere", service=service, events=events) self.store.get_appservice_state = Mock( return_value=defer.succeed(ApplicationServiceState.DOWN) @@ -82,7 +75,7 @@ class ApplicationServiceSchedulerTransactionCtrlTestCase(unittest.TestCase): ) # actual call - self.txnctrl.start_polling() + self.txnctrl.send(service, events) self.store.create_appservice_txn.assert_called_once_with( service=service, events=events # txn made and saved @@ -90,18 +83,15 @@ class ApplicationServiceSchedulerTransactionCtrlTestCase(unittest.TestCase): self.assertEquals(0, txn.send.call_count) # txn not sent though self.assertEquals(0, txn.complete.call_count) # or completed - def test_poll_single_group_service_up(self): + def test_single_service_up_txn_not_sent(self): # Test: The AS is up and the txn is not sent. A Recoverer is made and # started. service = Mock() events = [Mock(), Mock()] - groups = {} - groups[service] = events txn_id = "foobar" txn = Mock(id=txn_id, service=service, events=events) # mock methods - self.event_grouper.drain_groups = Mock(return_value=groups) self.store.get_appservice_state = Mock( return_value=defer.succeed(ApplicationServiceState.UP) ) @@ -112,7 +102,7 @@ class ApplicationServiceSchedulerTransactionCtrlTestCase(unittest.TestCase): ) # actual call - self.txnctrl.start_polling() + self.txnctrl.send(service, events) self.store.create_appservice_txn.assert_called_once_with( service=service, events=events @@ -125,12 +115,6 @@ class ApplicationServiceSchedulerTransactionCtrlTestCase(unittest.TestCase): service, ApplicationServiceState.DOWN # service marked as down ) - def test_poll_no_groups(self): - self.as_api.push_bulk = Mock() - self.event_grouper.drain_groups = Mock(return_value={}) - self.txnctrl.start_polling() - self.assertEquals(0, self.as_api.push_bulk.call_count) - class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase): @@ -205,54 +189,24 @@ class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase): self.callback.assert_called_once_with(self.recoverer) -class ApplicationServiceSchedulerEventGrouperTestCase(unittest.TestCase): +class ApplicationServiceSchedulerQueuerTestCase(unittest.TestCase): def setUp(self): - self.grouper = _EventGrouper() - - def test_drain_single_event(self): - service = Mock() - event = Mock() - self.grouper.enqueue(service, event) - groups = self.grouper.drain_groups() - self.assertTrue(service in groups) - self.assertEquals([event], groups[service]) - self.assertEquals(1, len(groups.keys())) - # no more events - self.assertEquals(self.grouper.drain_groups(), {}) - - def test_drain_multiple_events(self): - service = Mock() - events = [Mock(), Mock(), Mock()] - for e in events: - self.grouper.enqueue(service, e) - groups = self.grouper.drain_groups() - self.assertTrue(service in groups) - self.assertEquals(events, groups[service]) - # no more events - self.assertEquals(self.grouper.drain_groups(), {}) - - def test_drain_multiple_services(self): - services = [Mock(), Mock(), Mock()] - events_a = [Mock(), Mock()] - events_b = [Mock()] - events_c = [Mock(), Mock(), Mock(), Mock()] - mappings = { - services[0]: events_a, - services[1]: events_b, - services[2]: events_c - } - for e in events_b: - self.grouper.enqueue(services[1], e) - for e in events_c: - self.grouper.enqueue(services[2], e) - for e in events_a: - self.grouper.enqueue(services[0], e) - - groups = self.grouper.drain_groups() - for service in services: - self.assertTrue(service in groups) - self.assertEquals(mappings[service], groups[service]) - self.assertEquals(3, len(groups.keys())) - # no more events - self.assertEquals(self.grouper.drain_groups(), {}) + self.txn_ctrl = Mock() + self.queuer = _ServiceQueuer(self.txn_ctrl) + + def test_send_single_event_no_queue(self): + # Expect the event to be sent immediately. + pass + + def test_send_single_event_with_queue(self): + # - Send an event and don't resolve it just yet. + # - Send another event: expect send() to NOT be called. + # - Resolve the send event + # - Expect queued event to be sent + pass + + def test_multiple_service_queues(self): + # Tests that each service has its own queue, and that they don't block + # on each other. + pass -- cgit 1.5.1 From d04fa1f7121d996e05bd4def14951d89eb47d1ab Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Mon, 16 Mar 2015 14:03:16 +0000 Subject: Implement ServiceQueuer with tests. --- synapse/appservice/scheduler.py | 46 +++++++++++++++++++++----------- tests/appservice/test_scheduler.py | 54 +++++++++++++++++++++++++++++++++----- 2 files changed, 77 insertions(+), 23 deletions(-) (limited to 'tests') diff --git a/synapse/appservice/scheduler.py b/synapse/appservice/scheduler.py index 54c42d1b94..3cedd479a2 100644 --- a/synapse/appservice/scheduler.py +++ b/synapse/appservice/scheduler.py @@ -83,7 +83,6 @@ class AppServiceScheduler(object): self.clock, self.store, self.as_api, self.txn_ctrl.on_recovered ) self.txn_ctrl.add_recoverers(recoverers) - self.txn_ctrl.start_polling() def submit_event_for_as(self, service, event): self.queuer.enqueue(service, event) @@ -96,24 +95,37 @@ class _ServiceQueuer(object): """ def __init__(self, txn_ctrl): - self.groups = {} # dict of {service: [events]} + self.queued_events = {} # dict of {service_id: [events]} + self.pending_requests = {} # dict of {service_id: Deferred} self.txn_ctrl = txn_ctrl def enqueue(self, service, event): - # if nothing in queue for this service, send event immediately and add - # callbacks. - self.txn_ctrl.send(service, [event]) - - # else add to queue for this service - - if service not in self.groups: - self.groups[service] = [] - self.groups[service].append(event) - - def drain_groups(self): - groups = self.groups - self.groups = {} - return groups + # if this service isn't being sent something + if not self.pending_requests.get(service.id): + self._send_request(service, [event]) + else: + # add to queue for this service + if service.id not in self.queued_events: + self.queued_events[service.id] = [] + self.queued_events[service.id].append(event) + + def _send_request(self, service, events): + # send request and add callbacks + d = self.txn_ctrl.send(service, events) + d.addCallback(self._on_request_finish) + d.addErrback(self._on_request_fail) + self.pending_requests[service.id] = d + + def _on_request_finish(self, service): + self.pending_requests[service.id] = None + # if there are queued events, then send them. + if (service.id in self.queued_events + and len(self.queued_events[service.id]) > 0): + self._send_request(service, self.queued_events[service.id]) + self.queued_events[service.id] = [] + + def _on_request_fail(self, err): + logger.error("AS request failed: %s", err) class _TransactionController(object): @@ -142,6 +154,8 @@ class _TransactionController(object): self._start_recoverer(service) except Exception as e: logger.exception(e) + # request has finished + defer.returnValue(service) @defer.inlineCallbacks def on_recovered(self, recoverer): diff --git a/tests/appservice/test_scheduler.py b/tests/appservice/test_scheduler.py index 38d792eb02..82a5965097 100644 --- a/tests/appservice/test_scheduler.py +++ b/tests/appservice/test_scheduler.py @@ -197,16 +197,56 @@ class ApplicationServiceSchedulerQueuerTestCase(unittest.TestCase): def test_send_single_event_no_queue(self): # Expect the event to be sent immediately. - pass + service = Mock(id=4) + event = Mock() + self.queuer.enqueue(service, event) + self.txn_ctrl.send.assert_called_once_with(service, [event]) def test_send_single_event_with_queue(self): - # - Send an event and don't resolve it just yet. - # - Send another event: expect send() to NOT be called. - # - Resolve the send event - # - Expect queued event to be sent - pass + d = defer.Deferred() + self.txn_ctrl.send = Mock(return_value=d) + service = Mock(id=4) + event = Mock(event_id="first") + event2 = Mock(event_id="second") + event3 = Mock(event_id="third") + # Send an event and don't resolve it just yet. + self.queuer.enqueue(service, event) + # Send more events: expect send() to NOT be called multiple times. + self.queuer.enqueue(service, event2) + self.queuer.enqueue(service, event3) + self.txn_ctrl.send.assert_called_with(service, [event]) + self.assertEquals(1, self.txn_ctrl.send.call_count) + # Resolve the send event: expect the queued events to be sent + d.callback(service) + self.txn_ctrl.send.assert_called_with(service, [event2, event3]) + self.assertEquals(2, self.txn_ctrl.send.call_count) def test_multiple_service_queues(self): # Tests that each service has its own queue, and that they don't block # on each other. - pass + srv1 = Mock(id=4) + srv_1_defer = defer.Deferred() + srv_1_event = Mock(event_id="srv1a") + srv_1_event2 = Mock(event_id="srv1b") + + srv2 = Mock(id=6) + srv_2_defer = defer.Deferred() + srv_2_event = Mock(event_id="srv2a") + srv_2_event2 = Mock(event_id="srv2b") + + send_return_list = [srv_1_defer, srv_2_defer] + self.txn_ctrl.send = Mock(side_effect=lambda x,y: send_return_list.pop(0)) + + # send events for different ASes and make sure they are sent + self.queuer.enqueue(srv1, srv_1_event) + self.queuer.enqueue(srv1, srv_1_event2) + self.txn_ctrl.send.assert_called_with(srv1, [srv_1_event]) + self.queuer.enqueue(srv2, srv_2_event) + self.queuer.enqueue(srv2, srv_2_event2) + self.txn_ctrl.send.assert_called_with(srv2, [srv_2_event]) + + # make sure callbacks for a service only send queued events for THAT + # service + srv_2_defer.callback(srv2) + self.txn_ctrl.send.assert_called_with(srv2, [srv_2_event2]) + self.assertEquals(3, self.txn_ctrl.send.call_count) -- cgit 1.5.1 From 7e282a53a542832f44119b886b2d4d474bbe1d0f Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 20 Mar 2015 14:59:48 +0000 Subject: Tidy up _simple_... methods --- synapse/storage/_base.py | 99 +++++++++++++++++++++++++++------------------ synapse/storage/presence.py | 1 - tests/storage/test_base.py | 2 +- tests/storage/test_room.py | 2 +- 4 files changed, 62 insertions(+), 42 deletions(-) (limited to 'tests') diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 0260b4e645..a6f94de01e 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -399,7 +399,7 @@ class SQLBaseStore(object): txn.execute(sql, allvalues.values()) def _simple_select_one(self, table, keyvalues, retcols, - allow_none=False): + allow_none=False, desc="_simple_select_one"): """Executes a SELECT query on the named table, which is expected to return a single row, returning a single column from it. @@ -411,8 +411,10 @@ class SQLBaseStore(object): allow_none : If true, return None instead of failing if the SELECT statement returns no rows """ - return self._simple_selectupdate_one( - table, keyvalues, retcols=retcols, allow_none=allow_none + return self.runInteraction( + desc, + self._simple_select_one_txn, + table, keyvalues, retcols, allow_none, ) def _simple_select_one_onecol(self, table, keyvalues, retcol, @@ -523,7 +525,7 @@ class SQLBaseStore(object): return self.cursor_to_dict(txn) def _simple_update_one(self, table, keyvalues, updatevalues, - retcols=None): + desc="_simple_update_one"): """Executes an UPDATE query on the named table, setting new values for columns in a row matching the key values. @@ -541,52 +543,71 @@ class SQLBaseStore(object): get-and-set. This can be used to implement compare-and-set by putting the update column in the 'keyvalues' dict as well. """ - return self._simple_selectupdate_one(table, keyvalues, updatevalues, - retcols=retcols) + return self.runInteraction( + desc, + self._simple_update_one_txn, + table, keyvalues, updatevalues, + ) + + def _simple_update_one_txn(self, txn, table, keyvalues, updatevalues): + update_sql = "UPDATE %s SET %s WHERE %s" % ( + table, + ", ".join("%s = ?" % (k,) for k in updatevalues), + " AND ".join("%s = ?" % (k,) for k in keyvalues) + ) + + txn.execute( + update_sql, + updatevalues.values() + keyvalues.values() + ) + + if txn.rowcount == 0: + raise StoreError(404, "No row found") + if txn.rowcount > 1: + raise StoreError(500, "More than one row matched") + + def _simple_select_one_txn(self, txn, table, keyvalues, retcols, + allow_none=False): + select_sql = "SELECT %s FROM %s WHERE %s ORDER BY rowid asc" % ( + ", ".join(retcols), + table, + " AND ".join("%s = ?" % (k) for k in keyvalues) + ) + + txn.execute(select_sql, keyvalues.values()) + + row = txn.fetchone() + if not row: + if allow_none: + return None + raise StoreError(404, "No row found") + if txn.rowcount > 1: + raise StoreError(500, "More than one row matched") + + return dict(zip(retcols, row)) def _simple_selectupdate_one(self, table, keyvalues, updatevalues=None, retcols=None, allow_none=False): """ Combined SELECT then UPDATE.""" - if retcols: - select_sql = "SELECT %s FROM %s WHERE %s ORDER BY rowid asc" % ( - ", ".join(retcols), - table, - " AND ".join("%s = ?" % (k) for k in keyvalues) - ) - - if updatevalues: - update_sql = "UPDATE %s SET %s WHERE %s" % ( - table, - ", ".join("%s = ?" % (k,) for k in updatevalues), - " AND ".join("%s = ?" % (k,) for k in keyvalues) - ) - def func(txn): ret = None if retcols: - txn.execute(select_sql, keyvalues.values()) - - row = txn.fetchone() - if not row: - if allow_none: - return None - raise StoreError(404, "No row found") - if txn.rowcount > 1: - raise StoreError(500, "More than one row matched") - - ret = dict(zip(retcols, row)) + ret = self._simple_select_one_txn( + txn, + table=table, + keyvalues=keyvalues, + retcols=retcols, + allow_none=allow_none, + ) if updatevalues: - txn.execute( - update_sql, - updatevalues.values() + keyvalues.values() + self._simple_update_one_txn( + txn, + table=table, + keyvalues=keyvalues, + updatevalues=updatevalues, ) - if txn.rowcount == 0: - raise StoreError(404, "No row found") - if txn.rowcount > 1: - raise StoreError(500, "More than one row matched") - return ret return self.runInteraction("_simple_selectupdate_one", func) diff --git a/synapse/storage/presence.py b/synapse/storage/presence.py index 1dcd34723b..0084d67e5b 100644 --- a/synapse/storage/presence.py +++ b/synapse/storage/presence.py @@ -45,7 +45,6 @@ class PresenceStore(SQLBaseStore): updatevalues={"state": new_state["state"], "status_msg": new_state["status_msg"], "mtime": self._clock.time_msec()}, - retcols=["state"], ) def allow_presence_visible(self, observed_localpart, observer_userid): diff --git a/tests/storage/test_base.py b/tests/storage/test_base.py index 55fbffa7a2..7f5845cf0c 100644 --- a/tests/storage/test_base.py +++ b/tests/storage/test_base.py @@ -180,7 +180,7 @@ class SQLBaseStoreTestCase(unittest.TestCase): self.mock_txn.rowcount = 1 self.mock_txn.fetchone.return_value = ("Old Value",) - ret = yield self.datastore._simple_update_one( + ret = yield self.datastore._simple_selectupdate_one( table="tablename", keyvalues={"keycol": "TheKey"}, updatevalues={"columname": "New Value"}, diff --git a/tests/storage/test_room.py b/tests/storage/test_room.py index c88dd446fb..ab7625a3ca 100644 --- a/tests/storage/test_room.py +++ b/tests/storage/test_room.py @@ -44,7 +44,7 @@ class RoomStoreTestCase(unittest.TestCase): @defer.inlineCallbacks def test_get_room(self): - self.assertObjectHasAttributes( + self.assertDictContainsSubset( {"room_id": self.room.to_string(), "creator": self.u_creator.to_string(), "is_public": True}, -- cgit 1.5.1 From 0f86312c4cb262ad1b69207dd46712707dee75bb Mon Sep 17 00:00:00 2001 From: "Paul \"LeoNerd\" Evans" Date: Fri, 20 Mar 2015 18:13:49 +0000 Subject: Pull out the cache logic from the @cached wrapper into its own class we can reuse --- synapse/storage/_base.py | 89 +++++++++++++++++++++++++++------------------ tests/storage/test__base.py | 34 ++++++++++++++++- 2 files changed, 87 insertions(+), 36 deletions(-) (limited to 'tests') diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 2552a74f85..27ea65a0f6 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -53,6 +53,47 @@ cache_counter = metrics.register_cache( ) +class Cache(object): + + def __init__(self, name, max_entries=1000, keylen=1): + self.cache = OrderedDict() + + self.max_entries = max_entries + self.name = name + self.keylen = keylen + + caches_by_name[name] = self.cache + + def get(self, *keyargs): + if len(keyargs) != self.keylen: + raise ValueError("Expected a key to have %d items", self.keylen) + + if keyargs in self.cache: + cache_counter.inc_hits(self.name) + return self.cache[keyargs] + + cache_counter.inc_misses(self.name) + raise KeyError() + + def prefill(self, *args): # because I can't *keyargs, value + keyargs = args[:-1] + value = args[-1] + + if len(keyargs) != self.keylen: + raise ValueError("Expected a key to have %d items", self.keylen) + + while len(self.cache) > self.max_entries: + self.cache.popitem(last=False) + + self.cache[keyargs] = value + + def invalidate(self, *keyargs): + if len(keyargs) != self.keylen: + raise ValueError("Expected a key to have %d items", self.keylen) + + self.cache.pop(keyargs, None) + + # TODO(paul): # * consider other eviction strategies - LRU? def cached(max_entries=1000, num_args=1): @@ -70,48 +111,26 @@ def cached(max_entries=1000, num_args=1): calling the calculation function. """ def wrap(orig): - cache = OrderedDict() - name = orig.__name__ - - caches_by_name[name] = cache - - def prefill(*args): # because I can't *keyargs, value - keyargs = args[:-1] - value = args[-1] - - if len(keyargs) != num_args: - raise ValueError("Expected a call to have %d arguments", num_args) - - while len(cache) > max_entries: - cache.popitem(last=False) - - cache[keyargs] = value + cache = Cache( + name=orig.__name__, + max_entries=max_entries, + keylen=num_args, + ) @functools.wraps(orig) @defer.inlineCallbacks def wrapped(self, *keyargs): - if len(keyargs) != num_args: - raise ValueError("Expected a call to have %d arguments", num_args) - - if keyargs in cache: - cache_counter.inc_hits(name) - defer.returnValue(cache[keyargs]) - - cache_counter.inc_misses(name) - ret = yield orig(self, *keyargs) - - prefill(*keyargs + (ret,)) - - defer.returnValue(ret) + try: + defer.returnValue(cache.get(*keyargs)) + except KeyError: + ret = yield orig(self, *keyargs) - def invalidate(*keyargs): - if len(keyargs) != num_args: - raise ValueError("Expected a call to have %d arguments", num_args) + cache.prefill(*keyargs + (ret,)) - cache.pop(keyargs, None) + defer.returnValue(ret) - wrapped.invalidate = invalidate - wrapped.prefill = prefill + wrapped.invalidate = cache.invalidate + wrapped.prefill = cache.prefill return wrapped return wrap diff --git a/tests/storage/test__base.py b/tests/storage/test__base.py index 55d22f665a..783abc2b00 100644 --- a/tests/storage/test__base.py +++ b/tests/storage/test__base.py @@ -17,7 +17,39 @@ from tests import unittest from twisted.internet import defer -from synapse.storage._base import cached +from synapse.storage._base import Cache, cached + + +class CacheTestCase(unittest.TestCase): + + def setUp(self): + self.cache = Cache("test") + + def test_empty(self): + failed = False + try: + self.cache.get("foo") + except KeyError: + failed = True + + self.assertTrue(failed) + + def test_hit(self): + self.cache.prefill("foo", 123) + + self.assertEquals(self.cache.get("foo"), 123) + + def test_invalidate(self): + self.cache.prefill("foo", 123) + self.cache.invalidate("foo") + + failed = False + try: + self.cache.get("foo") + except KeyError: + failed = True + + self.assertTrue(failed) class CacheDecoratorTestCase(unittest.TestCase): -- cgit 1.5.1 From 438a21c87bec6386be1a9ed92d61b1b467a40bcd Mon Sep 17 00:00:00 2001 From: David Baker Date: Tue, 24 Mar 2015 18:21:54 +0000 Subject: Don't test exact equality of the list: as long as it has the fields we expect, that's just fine. I added the user_id (as in database pkey) and it broke: no point testing what that comes out as: it's determined by the db. --- tests/storage/test_registration.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) (limited to 'tests') diff --git a/tests/storage/test_registration.py b/tests/storage/test_registration.py index e0b81f2b57..93be02f8c3 100644 --- a/tests/storage/test_registration.py +++ b/tests/storage/test_registration.py @@ -38,13 +38,12 @@ class RegistrationStoreTestCase(unittest.TestCase): @defer.inlineCallbacks def test_register(self): yield self.store.register(self.user_id, self.tokens[0], self.pwhash) + u = yield self.store.get_user_by_id(self.user_id)[0] - self.assertEquals( - # TODO(paul): Surely this field should be 'user_id', not 'name' - # Additionally surely it shouldn't come in a 1-element list - [{"name": self.user_id, "password_hash": self.pwhash}], - (yield self.store.get_user_by_id(self.user_id)) - ) + # TODO(paul): Surely this field should be 'user_id', not 'name' + # Additionally surely it shouldn't come in a 1-element list + self.assertEquals(self.user_id, u['name']) + self.assertEquals(self.pwhash, u['password_hash']) self.assertEquals( {"admin": 0, -- cgit 1.5.1 From ce2766d19c985d0100cc143477100c2723be1844 Mon Sep 17 00:00:00 2001 From: David Baker Date: Tue, 24 Mar 2015 18:56:51 +0000 Subject: Fix tests --- tests/rest/client/v1/test_presence.py | 6 +++--- tests/rest/client/v1/test_profile.py | 2 +- tests/rest/client/v1/test_rooms.py | 16 ++++++++-------- tests/rest/client/v1/test_typing.py | 2 +- tests/storage/test_registration.py | 2 +- 5 files changed, 14 insertions(+), 14 deletions(-) (limited to 'tests') diff --git a/tests/rest/client/v1/test_presence.py b/tests/rest/client/v1/test_presence.py index b9c03383a2..8e0c5fa630 100644 --- a/tests/rest/client/v1/test_presence.py +++ b/tests/rest/client/v1/test_presence.py @@ -75,7 +75,7 @@ class PresenceStateTestCase(unittest.TestCase): "token_id": 1, } - hs.get_auth().get_user_by_token = _get_user_by_token + hs.get_v1auth().get_user_by_token = _get_user_by_token room_member_handler = hs.handlers.room_member_handler = Mock( spec=[ @@ -170,7 +170,7 @@ class PresenceListTestCase(unittest.TestCase): ] ) - hs.get_auth().get_user_by_token = _get_user_by_token + hs.get_v1auth().get_user_by_token = _get_user_by_token presence.register_servlets(hs, self.mock_resource) @@ -277,7 +277,7 @@ class PresenceEventStreamTestCase(unittest.TestCase): def _get_user_by_req(req=None): return (UserID.from_string(myid), "") - hs.get_auth().get_user_by_req = _get_user_by_req + hs.get_v1auth().get_user_by_req = _get_user_by_req presence.register_servlets(hs, self.mock_resource) events.register_servlets(hs, self.mock_resource) diff --git a/tests/rest/client/v1/test_profile.py b/tests/rest/client/v1/test_profile.py index 5cd5767f2e..929e5e5dd4 100644 --- a/tests/rest/client/v1/test_profile.py +++ b/tests/rest/client/v1/test_profile.py @@ -55,7 +55,7 @@ class ProfileTestCase(unittest.TestCase): def _get_user_by_req(request=None): return (UserID.from_string(myid), "") - hs.get_auth().get_user_by_req = _get_user_by_req + hs.get_v1auth().get_user_by_req = _get_user_by_req hs.get_handlers().profile_handler = self.mock_handler diff --git a/tests/rest/client/v1/test_rooms.py b/tests/rest/client/v1/test_rooms.py index 72fb4576b1..c83348acf9 100644 --- a/tests/rest/client/v1/test_rooms.py +++ b/tests/rest/client/v1/test_rooms.py @@ -61,7 +61,7 @@ class RoomPermissionsTestCase(RestTestCase): "device_id": None, "token_id": 1, } - hs.get_auth().get_user_by_token = _get_user_by_token + hs.get_v1auth().get_user_by_token = _get_user_by_token def _insert_client_ip(*args, **kwargs): return defer.succeed(None) @@ -71,7 +71,7 @@ class RoomPermissionsTestCase(RestTestCase): synapse.rest.client.v1.room.register_servlets(hs, self.mock_resource) - self.auth = hs.get_auth() + self.auth = hs.get_v1auth() # create some rooms under the name rmcreator_id self.uncreated_rmid = "!aa:test" @@ -448,7 +448,7 @@ class RoomsMemberListTestCase(RestTestCase): "device_id": None, "token_id": 1, } - hs.get_auth().get_user_by_token = _get_user_by_token + hs.get_v1auth().get_user_by_token = _get_user_by_token def _insert_client_ip(*args, **kwargs): return defer.succeed(None) @@ -528,7 +528,7 @@ class RoomsCreateTestCase(RestTestCase): "device_id": None, "token_id": 1, } - hs.get_auth().get_user_by_token = _get_user_by_token + hs.get_v1auth().get_user_by_token = _get_user_by_token def _insert_client_ip(*args, **kwargs): return defer.succeed(None) @@ -622,7 +622,7 @@ class RoomTopicTestCase(RestTestCase): "token_id": 1, } - hs.get_auth().get_user_by_token = _get_user_by_token + hs.get_v1auth().get_user_by_token = _get_user_by_token def _insert_client_ip(*args, **kwargs): return defer.succeed(None) @@ -728,7 +728,7 @@ class RoomMemberStateTestCase(RestTestCase): "device_id": None, "token_id": 1, } - hs.get_auth().get_user_by_token = _get_user_by_token + hs.get_v1auth().get_user_by_token = _get_user_by_token def _insert_client_ip(*args, **kwargs): return defer.succeed(None) @@ -855,7 +855,7 @@ class RoomMessagesTestCase(RestTestCase): "device_id": None, "token_id": 1, } - hs.get_auth().get_user_by_token = _get_user_by_token + hs.get_v1auth().get_user_by_token = _get_user_by_token def _insert_client_ip(*args, **kwargs): return defer.succeed(None) @@ -952,7 +952,7 @@ class RoomInitialSyncTestCase(RestTestCase): "device_id": None, "token_id": 1, } - hs.get_auth().get_user_by_token = _get_user_by_token + hs.get_v1auth().get_user_by_token = _get_user_by_token def _insert_client_ip(*args, **kwargs): return defer.succeed(None) diff --git a/tests/rest/client/v1/test_typing.py b/tests/rest/client/v1/test_typing.py index 80f2ec9ddf..3442e6ede5 100644 --- a/tests/rest/client/v1/test_typing.py +++ b/tests/rest/client/v1/test_typing.py @@ -67,7 +67,7 @@ class RoomTypingTestCase(RestTestCase): "token_id": 1, } - hs.get_auth().get_user_by_token = _get_user_by_token + hs.get_v1auth().get_user_by_token = _get_user_by_token def _insert_client_ip(*args, **kwargs): return defer.succeed(None) diff --git a/tests/storage/test_registration.py b/tests/storage/test_registration.py index 93be02f8c3..2f8953f518 100644 --- a/tests/storage/test_registration.py +++ b/tests/storage/test_registration.py @@ -38,7 +38,7 @@ class RegistrationStoreTestCase(unittest.TestCase): @defer.inlineCallbacks def test_register(self): yield self.store.register(self.user_id, self.tokens[0], self.pwhash) - u = yield self.store.get_user_by_id(self.user_id)[0] + u = (yield self.store.get_user_by_id(self.user_id))[0] # TODO(paul): Surely this field should be 'user_id', not 'name' # Additionally surely it shouldn't come in a 1-element list -- cgit 1.5.1 From 7ab9f91a605d501cadee1c212eca2ef0467adc50 Mon Sep 17 00:00:00 2001 From: "Paul \"LeoNerd\" Evans" Date: Wed, 25 Mar 2015 18:50:43 +0000 Subject: Unit-test that Cache() key eviction is ordered --- tests/storage/test__base.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) (limited to 'tests') diff --git a/tests/storage/test__base.py b/tests/storage/test__base.py index 783abc2b00..b6853ba2d4 100644 --- a/tests/storage/test__base.py +++ b/tests/storage/test__base.py @@ -51,6 +51,24 @@ class CacheTestCase(unittest.TestCase): self.assertTrue(failed) + def test_eviction(self): + cache = Cache("test", max_entries=2) + + cache.prefill(1, "one") + cache.prefill(2, "two") + cache.prefill(3, "three") # 1 will be evicted + + failed = False + try: + cache.get(1) + except KeyError: + failed = True + + self.assertTrue(failed) + + cache.get(2) + cache.get(3) + class CacheDecoratorTestCase(unittest.TestCase): -- cgit 1.5.1 From 9ba6487b3fe985c4ec84b02d9804aea7e2df6c40 Mon Sep 17 00:00:00 2001 From: "Paul \"LeoNerd\" Evans" Date: Wed, 25 Mar 2015 19:05:34 +0000 Subject: Allow a choice of LRU behaviour for Cache() by using LruCache() or OrderedDict() --- synapse/storage/_base.py | 20 ++++++++++++-------- tests/storage/test__base.py | 22 ++++++++++++++++++++++ 2 files changed, 34 insertions(+), 8 deletions(-) (limited to 'tests') diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 27ea65a0f6..6fa63f052e 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -55,10 +55,14 @@ cache_counter = metrics.register_cache( class Cache(object): - def __init__(self, name, max_entries=1000, keylen=1): - self.cache = OrderedDict() + def __init__(self, name, max_entries=1000, keylen=1, lru=False): + if lru: + self.cache = LruCache(max_size=max_entries) + self.max_entries = None + else: + self.cache = OrderedDict() + self.max_entries = max_entries - self.max_entries = max_entries self.name = name self.keylen = keylen @@ -82,8 +86,9 @@ class Cache(object): if len(keyargs) != self.keylen: raise ValueError("Expected a key to have %d items", self.keylen) - while len(self.cache) > self.max_entries: - self.cache.popitem(last=False) + if self.max_entries is not None: + while len(self.cache) >= self.max_entries: + self.cache.popitem(last=False) self.cache[keyargs] = value @@ -94,9 +99,7 @@ class Cache(object): self.cache.pop(keyargs, None) -# TODO(paul): -# * consider other eviction strategies - LRU? -def cached(max_entries=1000, num_args=1): +def cached(max_entries=1000, num_args=1, lru=False): """ A method decorator that applies a memoizing cache around the function. The function is presumed to take zero or more arguments, which are used in @@ -115,6 +118,7 @@ def cached(max_entries=1000, num_args=1): name=orig.__name__, max_entries=max_entries, keylen=num_args, + lru=lru, ) @functools.wraps(orig) diff --git a/tests/storage/test__base.py b/tests/storage/test__base.py index b6853ba2d4..96caf8c4c1 100644 --- a/tests/storage/test__base.py +++ b/tests/storage/test__base.py @@ -69,6 +69,28 @@ class CacheTestCase(unittest.TestCase): cache.get(2) cache.get(3) + def test_eviction_lru(self): + cache = Cache("test", max_entries=2, lru=True) + + cache.prefill(1, "one") + cache.prefill(2, "two") + + # Now access 1 again, thus causing 2 to be least-recently used + cache.get(1) + + cache.prefill(3, "three") + + failed = False + try: + cache.get(2) + except KeyError: + failed = True + + self.assertTrue(failed) + + cache.get(1) + cache.get(3) + class CacheDecoratorTestCase(unittest.TestCase): -- cgit 1.5.1 From d33ae65efc14a18a8a690d39d6e9c81aaafa1062 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Tue, 31 Mar 2015 11:00:00 +0100 Subject: Remove more reg/unreg methods. Read config not database for cache. --- synapse/handlers/appservice.py | 37 ------- synapse/storage/appservice.py | 219 +++++++++++---------------------------- tests/storage/test_appservice.py | 39 ------- 3 files changed, 59 insertions(+), 236 deletions(-) (limited to 'tests') diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py index 58b5b60bb7..59cf15b037 100644 --- a/synapse/handlers/appservice.py +++ b/synapse/handlers/appservice.py @@ -16,10 +16,8 @@ from twisted.internet import defer from synapse.api.constants import EventTypes, Membership -from synapse.api.errors import Codes, StoreError, SynapseError from synapse.appservice import ApplicationService from synapse.types import UserID -import synapse.util.stringutils as stringutils import logging @@ -49,38 +47,6 @@ class ApplicationServicesHandler(object): self.scheduler = appservice_scheduler self.started_scheduler = False - @defer.inlineCallbacks - def register(self, app_service): - logger.info("Register -> %s", app_service) - # check the token is recognised - try: - stored_service = yield self.store.get_app_service_by_token( - app_service.token - ) - if not stored_service: - raise StoreError(404, "Application service not found") - app_service.id = stored_service.id - except StoreError: - raise SynapseError( - 403, "Unrecognised application services token. " - "Consult the home server admin.", - errcode=Codes.FORBIDDEN - ) - app_service.hs_token = self._generate_hs_token() - - # create a sender for this application service which is used when - # creating rooms, etc.. - account = yield self.hs.get_handlers().registration_handler.register() - app_service.sender = account[0] - - yield self.store.update_app_service(app_service) - defer.returnValue(app_service) - - @defer.inlineCallbacks - def unregister(self, token): - logger.info("Unregister as_token=%s", token) - yield self.store.unregister_app_service(token) - @defer.inlineCallbacks def notify_interested_services(self, event): """Notifies (pushes) all application services interested in this event. @@ -223,6 +189,3 @@ class ApplicationServicesHandler(object): exists = yield self.query_user_exists(user_id) defer.returnValue(exists) defer.returnValue(True) - - def _generate_hs_token(self): - return stringutils.random_string(24) diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py index 93304a745f..fe9372a7c6 100644 --- a/synapse/storage/appservice.py +++ b/synapse/storage/appservice.py @@ -13,12 +13,12 @@ # See the License for the specific language governing permissions and # limitations under the License. import logging +import yaml from simplejson import JSONDecodeError import simplejson as json from twisted.internet import defer from synapse.api.constants import Membership -from synapse.api.errors import StoreError from synapse.appservice import ApplicationService, AppServiceTransaction from synapse.storage.roommember import RoomsForUser from ._base import SQLBaseStore @@ -27,141 +27,18 @@ from ._base import SQLBaseStore logger = logging.getLogger(__name__) -def log_failure(failure): - logger.error("Failed to detect application services: %s", failure.value) - logger.error(failure.getTraceback()) - - class ApplicationServiceStore(SQLBaseStore): def __init__(self, hs): super(ApplicationServiceStore, self).__init__(hs) self.services_cache = [] - self.cache_defer = self._populate_appservice_cache() - self.cache_defer.addErrback(log_failure) - - @defer.inlineCallbacks - def unregister_app_service(self, token): - """Unregisters this service. - - This removes all AS specific regex and the base URL. The token is the - only thing preserved for future registration attempts. - """ - yield self.cache_defer # make sure the cache is ready - yield self.runInteraction( - "unregister_app_service", - self._unregister_app_service_txn, - token, - ) - # update cache TODO: Should this be in the txn? - for service in self.services_cache: - if service.token == token: - service.url = None - service.namespaces = None - service.hs_token = None - - def _unregister_app_service_txn(self, txn, token): - # kill the url to prevent pushes - txn.execute( - "UPDATE application_services SET url=NULL WHERE token=?", - (token,) - ) - - # cleanup regex - as_id = self._get_as_id_txn(txn, token) - if not as_id: - logger.warning( - "unregister_app_service_txn: Failed to find as_id for token=", - token - ) - return False - - txn.execute( - "DELETE FROM application_services_regex WHERE as_id=?", - (as_id,) - ) - return True - - @defer.inlineCallbacks - def update_app_service(self, service): - """Update an application service, clobbering what was previously there. - - Args: - service(ApplicationService): The updated service. - """ - yield self.cache_defer # make sure the cache is ready - - # NB: There is no "insert" since we provide no public-facing API to - # allocate new ASes. It relies on the server admin inserting the AS - # token into the database manually. - - if not service.token or not service.url: - raise StoreError(400, "Token and url must be specified.") - - if not service.hs_token: - raise StoreError(500, "No HS token") - - as_id = yield self.runInteraction( - "update_app_service", - self._update_app_service_txn, - service + self._populate_appservice_cache( + hs.config.app_service_config_files ) - service.id = as_id - # update cache TODO: Should this be in the txn? - for (index, cache_service) in enumerate(self.services_cache): - if service.token == cache_service.token: - self.services_cache[index] = service - logger.info("Updated: %s", service) - return - # new entry - self.services_cache.append(service) - logger.info("Updated(new): %s", service) - - def _update_app_service_txn(self, txn, service): - as_id = self._get_as_id_txn(txn, service.token) - if not as_id: - logger.warning( - "update_app_service_txn: Failed to find as_id for token=", - service.token - ) - return - - txn.execute( - "UPDATE application_services SET url=?, hs_token=?, sender=? " - "WHERE id=?", - (service.url, service.hs_token, service.sender, as_id,) - ) - # cleanup regex - txn.execute( - "DELETE FROM application_services_regex WHERE as_id=?", - (as_id,) - ) - for (ns_int, ns_str) in enumerate(ApplicationService.NS_LIST): - if ns_str in service.namespaces: - for regex_obj in service.namespaces[ns_str]: - txn.execute( - "INSERT INTO application_services_regex(" - "as_id, namespace, regex) values(?,?,?)", - (as_id, ns_int, json.dumps(regex_obj)) - ) - return as_id - - def _get_as_id_txn(self, txn, token): - cursor = txn.execute( - "SELECT id FROM application_services WHERE token=?", - (token,) - ) - res = cursor.fetchone() - if res: - return res[0] - - @defer.inlineCallbacks def get_app_services(self): - yield self.cache_defer # make sure the cache is ready - defer.returnValue(self.services_cache) + defer.succeed(self.services_cache) - @defer.inlineCallbacks def get_app_service_by_user_id(self, user_id): """Retrieve an application service from their user ID. @@ -175,37 +52,24 @@ class ApplicationServiceStore(SQLBaseStore): Returns: synapse.appservice.ApplicationService or None. """ - - yield self.cache_defer # make sure the cache is ready - for service in self.services_cache: if service.sender == user_id: - defer.returnValue(service) + defer.succeed(service) return - defer.returnValue(None) + defer.succeed(None) - @defer.inlineCallbacks - def get_app_service_by_token(self, token, from_cache=True): + def get_app_service_by_token(self, token): """Get the application service with the given appservice token. Args: token (str): The application service token. - from_cache (bool): True to get this service from the cache, False to - check the database. - Raises: - StoreError if there was a problem retrieving this service. + Returns: + synapse.appservice.ApplicationService or None. """ - yield self.cache_defer # make sure the cache is ready - - if from_cache: - for service in self.services_cache: - if service.token == token: - defer.returnValue(service) - return - defer.returnValue(None) - - # TODO: The from_cache=False impl - # TODO: This should be JOINed with the application_services_regex table. + for service in self.services_cache: + if service.token == token: + return defer.succeed(service) + defer.succeed(None) def get_app_service_rooms(self, service): """Get a list of RoomsForUser for this application service. @@ -336,18 +200,53 @@ class ApplicationServiceStore(SQLBaseStore): )) return service_list - @defer.inlineCallbacks - def _populate_appservice_cache(self): - """Populates the ApplicationServiceCache from the database.""" - sql = ("SELECT r.*, a.* FROM application_services AS a LEFT JOIN " - "application_services_regex AS r ON a.id = r.as_id") - - results = yield self._execute_and_decode("appservice_cache", sql) - services = self._parse_services_dict(results) + def _load_appservice(self, as_info): + required_string_fields = ["url", "as_token", "hs_token", "sender"] + for field in required_string_fields: + if not isinstance(as_info.get(field), basestring): + raise KeyError("Required string field: '%s'", field) + + # namespace checks + if not isinstance(as_info.get("namespaces"), dict): + raise KeyError("Requires 'namespaces' object.") + for ns in ApplicationService.NS_LIST: + # specific namespaces are optional + if ns in as_info["namespaces"]: + # expect a list of dicts with exclusive and regex keys + for regex_obj in as_info["namespaces"][ns]: + if not isinstance(regex_obj, dict): + raise ValueError( + "Expected namespace entry in %s to be an object," + " but got %s", ns, regex_obj + ) + if not isinstance(regex_obj.get("regex"), basestring): + raise ValueError( + "Missing/bad type 'regex' key in %s", regex_obj + ) + if not isinstance(regex_obj.get("exclusive"), bool): + raise ValueError( + "Missing/bad type 'exclusive' key in %s", regex_obj + ) + return ApplicationService( + token=as_info["as_token"], + url=as_info["url"], + namespaces=as_info["namespaces"], + hs_token=as_info["hs_token"], + sender=as_info["sender"] + ) - for service in services: - logger.info("Found application service: %s", service) - self.services_cache.append(service) + def _populate_appservice_cache(self, config_files): + """Populates a cache of Application Services from the config files.""" + for config_file in config_files: + try: + with open(config_file, 'r') as f: + as_info = yaml.load(f) + appservice = self._load_appservice(as_info) + logger.info("Loaded application service: %s", appservice) + self.services_cache.append(appservice) + except Exception as e: + logger.error("Failed to load appservice from '%s'", config_file) + logger.exception(e) class ApplicationServiceTransactionStore(SQLBaseStore): diff --git a/tests/storage/test_appservice.py b/tests/storage/test_appservice.py index e79599f7fb..82bfea15a6 100644 --- a/tests/storage/test_appservice.py +++ b/tests/storage/test_appservice.py @@ -49,45 +49,6 @@ class ApplicationServiceStoreTestCase(unittest.TestCase): # must be done after inserts self.store = ApplicationServiceStore(hs) - @defer.inlineCallbacks - def test_update_and_retrieval_of_service(self): - url = "https://matrix.org/appservices/foobar" - hs_token = "hstok" - user_regex = [ - {"regex": "@foobar_.*:matrix.org", "exclusive": True} - ] - alias_regex = [ - {"regex": "#foobar_.*:matrix.org", "exclusive": False} - ] - room_regex = [ - - ] - service = ApplicationService( - url=url, hs_token=hs_token, token=self.as_token, namespaces={ - ApplicationService.NS_USERS: user_regex, - ApplicationService.NS_ALIASES: alias_regex, - ApplicationService.NS_ROOMS: room_regex - }) - yield self.store.update_app_service(service) - - stored_service = yield self.store.get_app_service_by_token( - self.as_token - ) - self.assertEquals(stored_service.token, self.as_token) - self.assertEquals(stored_service.url, url) - self.assertEquals( - stored_service.namespaces[ApplicationService.NS_ALIASES], - alias_regex - ) - self.assertEquals( - stored_service.namespaces[ApplicationService.NS_ROOMS], - room_regex - ) - self.assertEquals( - stored_service.namespaces[ApplicationService.NS_USERS], - user_regex - ) - @defer.inlineCallbacks def test_retrieve_unknown_service_token(self): service = yield self.store.get_app_service_by_token("invalid_token") -- cgit 1.5.1 From b59aa745560608c8503421bd9542c99fc1c571b5 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Tue, 31 Mar 2015 11:35:45 +0100 Subject: Fix tests and missing returns on deferreds. --- synapse/appservice/__init__.py | 2 +- synapse/storage/appservice.py | 18 +++++++++++------- tests/storage/test_appservice.py | 41 ++++++++++++++++++++++++++++------------ 3 files changed, 41 insertions(+), 20 deletions(-) (limited to 'tests') diff --git a/synapse/appservice/__init__.py b/synapse/appservice/__init__.py index c60db16b74..a8108c1efb 100644 --- a/synapse/appservice/__init__.py +++ b/synapse/appservice/__init__.py @@ -95,7 +95,7 @@ class ApplicationService(object): # rooms: [ {regex: "[A-z]+.*", exclusive: true}, ...], # } if not namespaces: - return None + namespaces = {} for ns in ApplicationService.NS_LIST: if ns not in namespaces: diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py index fe9372a7c6..a520a859d3 100644 --- a/synapse/storage/appservice.py +++ b/synapse/storage/appservice.py @@ -37,7 +37,7 @@ class ApplicationServiceStore(SQLBaseStore): ) def get_app_services(self): - defer.succeed(self.services_cache) + return defer.succeed(self.services_cache) def get_app_service_by_user_id(self, user_id): """Retrieve an application service from their user ID. @@ -54,9 +54,8 @@ class ApplicationServiceStore(SQLBaseStore): """ for service in self.services_cache: if service.sender == user_id: - defer.succeed(service) - return - defer.succeed(None) + return defer.succeed(service) + return defer.succeed(None) def get_app_service_by_token(self, token): """Get the application service with the given appservice token. @@ -69,7 +68,7 @@ class ApplicationServiceStore(SQLBaseStore): for service in self.services_cache: if service.token == token: return defer.succeed(service) - defer.succeed(None) + return defer.succeed(None) def get_app_service_rooms(self, service): """Get a list of RoomsForUser for this application service. @@ -237,11 +236,16 @@ class ApplicationServiceStore(SQLBaseStore): def _populate_appservice_cache(self, config_files): """Populates a cache of Application Services from the config files.""" + if not isinstance(config_files, list): + logger.warning( + "Expected %s to be a list of AS config files.", config_files + ) + return + for config_file in config_files: try: with open(config_file, 'r') as f: - as_info = yaml.load(f) - appservice = self._load_appservice(as_info) + appservice = self._load_appservice(yaml.load(f)) logger.info("Loaded application service: %s", appservice) self.services_cache.append(appservice) except Exception as e: diff --git a/tests/storage/test_appservice.py b/tests/storage/test_appservice.py index 82bfea15a6..b856438fd2 100644 --- a/tests/storage/test_appservice.py +++ b/tests/storage/test_appservice.py @@ -22,6 +22,8 @@ from synapse.storage.appservice import ( ) import json +import os +import yaml from mock import Mock from tests.utils import SQLiteMemoryDbPool, MockClock @@ -30,25 +32,40 @@ class ApplicationServiceStoreTestCase(unittest.TestCase): @defer.inlineCallbacks def setUp(self): + self.as_yaml_files = [] db_pool = SQLiteMemoryDbPool() yield db_pool.prepare() hs = HomeServer( - "test", db_pool=db_pool, clock=MockClock(), config=Mock() + "test", db_pool=db_pool, clock=MockClock(), + config=Mock( + app_service_config_files=self.as_yaml_files + ) ) + self.as_token = "token1" - db_pool.runQuery( - "INSERT INTO application_services(token) VALUES(?)", - (self.as_token,) - ) - db_pool.runQuery( - "INSERT INTO application_services(token) VALUES(?)", ("token2",) - ) - db_pool.runQuery( - "INSERT INTO application_services(token) VALUES(?)", ("token3",) - ) + self.as_url = "some_url" + self._add_appservice(self.as_token, self.as_url, "some_hs_token", "bob") + self._add_appservice("token2", "some_url", "some_hs_token", "bob") + self._add_appservice("token3", "some_url", "some_hs_token", "bob") # must be done after inserts self.store = ApplicationServiceStore(hs) + def tearDown(self): + # TODO: suboptimal that we need to create files for tests! + for f in self.as_yaml_files: + try: + os.remove(f) + except: + pass + + def _add_appservice(self, as_token, url, hs_token, sender): + as_yaml = dict(url=url, as_token=as_token, hs_token=hs_token, + sender=sender, namespaces={}) + # use the token as the filename + with open(as_token, 'w') as outfile: + outfile.write(yaml.dump(as_yaml)) + self.as_yaml_files.append(as_token) + @defer.inlineCallbacks def test_retrieve_unknown_service_token(self): service = yield self.store.get_app_service_by_token("invalid_token") @@ -60,7 +77,7 @@ class ApplicationServiceStoreTestCase(unittest.TestCase): self.as_token ) self.assertEquals(stored_service.token, self.as_token) - self.assertEquals(stored_service.url, None) + self.assertEquals(stored_service.url, self.as_url) self.assertEquals( stored_service.namespaces[ApplicationService.NS_ALIASES], [] -- cgit 1.5.1 From c217504949a90712f41a0422215f923b4d114a17 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Tue, 31 Mar 2015 12:07:56 +0100 Subject: Edit SQL schema to use string IDs not ints. Use token as ID. Update tests. --- synapse/storage/appservice.py | 23 ++++++++----- .../storage/schema/delta/15/appservice_txns.sql | 7 ++-- tests/storage/test_appservice.py | 38 +++++++++++++--------- 3 files changed, 40 insertions(+), 28 deletions(-) (limited to 'tests') diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py index a520a859d3..a8780eca1e 100644 --- a/synapse/storage/appservice.py +++ b/synapse/storage/appservice.py @@ -231,7 +231,8 @@ class ApplicationServiceStore(SQLBaseStore): url=as_info["url"], namespaces=as_info["namespaces"], hs_token=as_info["hs_token"], - sender=as_info["sender"] + sender=as_info["sender"], + id=as_info["as_token"] # the token is the only unique thing here ) def _populate_appservice_cache(self, config_files): @@ -268,16 +269,20 @@ class ApplicationServiceTransactionStore(SQLBaseStore): A Deferred which resolves to a list of ApplicationServices, which may be empty. """ - sql = ( - "SELECT r.*, a.* FROM application_services_state AS s LEFT JOIN" - " application_services AS a ON a.id=s.as_id LEFT JOIN" - " application_services_regex AS r ON r.as_id=a.id WHERE state = ?" - ) - results = yield self._execute_and_decode( - "get_appservices_by_state", sql, state + results = yield self._simple_select_list( + "application_services_state", + dict(state=state), + ["as_id"] ) # NB: This assumes this class is linked with ApplicationServiceStore - defer.returnValue(self._parse_services_dict(results)) + as_list = yield self.get_app_services() + services = [] + + for res in results: + for service in as_list: + if service.id == res["as_id"]: + services.append(service) + defer.returnValue(services) @defer.inlineCallbacks def get_appservice_state(self, service): diff --git a/synapse/storage/schema/delta/15/appservice_txns.sql b/synapse/storage/schema/delta/15/appservice_txns.sql index 13bbb2de2e..2b27e2a429 100644 --- a/synapse/storage/schema/delta/15/appservice_txns.sql +++ b/synapse/storage/schema/delta/15/appservice_txns.sql @@ -14,14 +14,13 @@ */ CREATE TABLE IF NOT EXISTS application_services_state( - as_id INTEGER PRIMARY KEY, + as_id TEXT PRIMARY KEY, state TEXT, - last_txn TEXT, - FOREIGN KEY(as_id) REFERENCES application_services(id) + last_txn TEXT ); CREATE TABLE IF NOT EXISTS application_services_txns( - as_id INTEGER NOT NULL, + as_id TEXT NOT NULL, txn_id INTEGER NOT NULL, event_ids TEXT NOT NULL, UNIQUE(as_id, txn_id) ON CONFLICT ROLLBACK diff --git a/tests/storage/test_appservice.py b/tests/storage/test_appservice.py index b856438fd2..58551e40b9 100644 --- a/tests/storage/test_appservice.py +++ b/tests/storage/test_appservice.py @@ -101,42 +101,48 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase): @defer.inlineCallbacks def setUp(self): + self.as_yaml_files = [] self.db_pool = SQLiteMemoryDbPool() yield self.db_pool.prepare() - hs = HomeServer( - "test", db_pool=self.db_pool, clock=MockClock(), config=Mock() - ) self.as_list = [ { "token": "token1", "url": "https://matrix-as.org", - "id": 3 + "id": "token1" }, { "token": "alpha_tok", "url": "https://alpha.com", - "id": 5 + "id": "alpha_tok" }, { "token": "beta_tok", "url": "https://beta.com", - "id": 6 + "id": "beta_tok" }, { "token": "delta_tok", "url": "https://delta.com", - "id": 7 + "id": "delta_tok" }, ] for s in self.as_list: - yield self._add_service(s["id"], s["url"], s["token"]) - self.store = TestTransactionStore(hs) + yield self._add_service(s["url"], s["token"]) - def _add_service(self, as_id, url, token): - return self.db_pool.runQuery( - "INSERT INTO application_services(id, url, token) VALUES(?,?,?)", - (as_id, url, token) + hs = HomeServer( + "test", db_pool=self.db_pool, clock=MockClock(), config=Mock( + app_service_config_files=self.as_yaml_files + ) ) + self.store = TestTransactionStore(hs) + + def _add_service(self, url, as_token): + as_yaml = dict(url=url, as_token=as_token, hs_token="something", + sender="a_sender", namespaces={}) + # use the token as the filename + with open(as_token, 'w') as outfile: + outfile.write(yaml.dump(as_yaml)) + self.as_yaml_files.append(as_token) def _set_state(self, id, state, txn=None): return self.db_pool.runQuery( @@ -388,8 +394,10 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase): ApplicationServiceState.DOWN ) self.assertEquals(2, len(services)) - self.assertEquals(self.as_list[2]["id"], services[0].id) - self.assertEquals(self.as_list[0]["id"], services[1].id) + self.assertEquals( + set([self.as_list[2]["id"], self.as_list[0]["id"]]), + set([services[0].id, services[1].id]) + ) # required for ApplicationServiceTransactionStoreTestCase tests -- cgit 1.5.1 From cf1fa59f4b72dbf5c9d735eaf051f1456721d91f Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Tue, 31 Mar 2015 13:48:03 +0100 Subject: Use a sender localpart instead of a user ID. Form the user ID at runtime instead, This gives less room for error in AS config files since they cannot specify the domain of another HS. --- synapse/storage/appservice.py | 11 +++++++++-- tests/storage/test_appservice.py | 4 ++-- 2 files changed, 11 insertions(+), 4 deletions(-) (limited to 'tests') diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py index a8780eca1e..557e377ca5 100644 --- a/synapse/storage/appservice.py +++ b/synapse/storage/appservice.py @@ -21,6 +21,7 @@ from twisted.internet import defer from synapse.api.constants import Membership from synapse.appservice import ApplicationService, AppServiceTransaction from synapse.storage.roommember import RoomsForUser +from synapse.types import UserID from ._base import SQLBaseStore @@ -31,6 +32,7 @@ class ApplicationServiceStore(SQLBaseStore): def __init__(self, hs): super(ApplicationServiceStore, self).__init__(hs) + self.hostname = hs.hostname self.services_cache = [] self._populate_appservice_cache( hs.config.app_service_config_files @@ -200,11 +202,16 @@ class ApplicationServiceStore(SQLBaseStore): return service_list def _load_appservice(self, as_info): - required_string_fields = ["url", "as_token", "hs_token", "sender"] + required_string_fields = [ + "url", "as_token", "hs_token", "sender_localpart" + ] for field in required_string_fields: if not isinstance(as_info.get(field), basestring): raise KeyError("Required string field: '%s'", field) + user = UserID(as_info["sender_localpart"], self.hostname) + user_id = user.to_string() + # namespace checks if not isinstance(as_info.get("namespaces"), dict): raise KeyError("Requires 'namespaces' object.") @@ -231,7 +238,7 @@ class ApplicationServiceStore(SQLBaseStore): url=as_info["url"], namespaces=as_info["namespaces"], hs_token=as_info["hs_token"], - sender=as_info["sender"], + sender=user_id, id=as_info["as_token"] # the token is the only unique thing here ) diff --git a/tests/storage/test_appservice.py b/tests/storage/test_appservice.py index 58551e40b9..675959c56c 100644 --- a/tests/storage/test_appservice.py +++ b/tests/storage/test_appservice.py @@ -60,7 +60,7 @@ class ApplicationServiceStoreTestCase(unittest.TestCase): def _add_appservice(self, as_token, url, hs_token, sender): as_yaml = dict(url=url, as_token=as_token, hs_token=hs_token, - sender=sender, namespaces={}) + sender_localpart=sender, namespaces={}) # use the token as the filename with open(as_token, 'w') as outfile: outfile.write(yaml.dump(as_yaml)) @@ -138,7 +138,7 @@ class ApplicationServiceTransactionStoreTestCase(unittest.TestCase): def _add_service(self, url, as_token): as_yaml = dict(url=url, as_token=as_token, hs_token="something", - sender="a_sender", namespaces={}) + sender_localpart="a_sender", namespaces={}) # use the token as the filename with open(as_token, 'w') as outfile: outfile.write(yaml.dump(as_yaml)) -- cgit 1.5.1 From 09cbff174a01757d10107b7960972a484153323e Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Tue, 31 Mar 2015 16:43:49 +0100 Subject: Fix thinko whereby events *for the AS specifically* were not passed on. This was caused by not explicitly checking the service.sender field. This has now been fixed and a regression test has been added. --- synapse/appservice/__init__.py | 5 ++++- tests/appservice/test_appservice.py | 13 +++++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) (limited to 'tests') diff --git a/synapse/appservice/__init__.py b/synapse/appservice/__init__.py index c60db16b74..4a6cdbc2be 100644 --- a/synapse/appservice/__init__.py +++ b/synapse/appservice/__init__.py @@ -199,7 +199,10 @@ class ApplicationService(object): return self._matches_user(event, member_list) def is_interested_in_user(self, user_id): - return self._matches_regex(user_id, ApplicationService.NS_USERS) + return ( + self._matches_regex(user_id, ApplicationService.NS_USERS) + or user_id == self.sender + ) def is_interested_in_alias(self, alias): return self._matches_regex(alias, ApplicationService.NS_ALIASES) diff --git a/tests/appservice/test_appservice.py b/tests/appservice/test_appservice.py index eb7becf725..62149d6902 100644 --- a/tests/appservice/test_appservice.py +++ b/tests/appservice/test_appservice.py @@ -199,6 +199,19 @@ class ApplicationServiceTestCase(unittest.TestCase): aliases_for_event=["#xmpp_barfoo:matrix.org"] )) + def test_interested_in_self(self): + # make sure invites get through + self.service.sender = "@appservice:name" + self.service.namespaces[ApplicationService.NS_USERS].append( + _regex("@irc_.*") + ) + self.event.type = "m.room.member" + self.event.content = { + "membership": "invite" + } + self.event.state_key = self.service.sender + self.assertTrue(self.service.is_interested(self.event)) + def test_member_list_match(self): self.service.namespaces[ApplicationService.NS_USERS].append( _regex("@irc_.*") -- cgit 1.5.1 From 9236136f3a4f0d8119d4a6333f37378f8e259e4a Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 1 Apr 2015 14:12:33 +0100 Subject: Make work in both Maria and SQLite. Fix tests --- synapse/app/homeserver.py | 20 ++++++---- synapse/storage/__init__.py | 43 +++++++++++----------- synapse/storage/_base.py | 30 ++++++++++----- synapse/storage/engines/__init__.py | 35 ++++++++++++++++++ synapse/storage/engines/maria.py | 30 +++++++++++++++ synapse/storage/engines/sqlite3.py | 25 +++++++++++++ synapse/storage/keys.py | 4 +- synapse/storage/registration.py | 8 +++- synapse/storage/schema/delta/12/v12.sql | 8 ++-- synapse/storage/schema/delta/13/v13.sql | 4 +- synapse/storage/schema/delta/14/v14.sql | 2 +- .../storage/schema/full_schemas/11/event_edges.sql | 14 +++---- .../schema/full_schemas/11/event_signatures.sql | 8 ++-- synapse/storage/schema/full_schemas/11/im.sql | 18 ++++----- synapse/storage/schema/full_schemas/11/keys.sql | 4 +- .../schema/full_schemas/11/media_repository.sql | 8 ++-- .../storage/schema/full_schemas/11/presence.sql | 6 +-- .../storage/schema/full_schemas/11/profiles.sql | 4 +- .../storage/schema/full_schemas/11/redactions.sql | 2 +- .../schema/full_schemas/11/room_aliases.sql | 8 ++-- synapse/storage/schema/full_schemas/11/state.sql | 6 +-- .../schema/full_schemas/11/transactions.sql | 8 ++-- synapse/storage/schema/full_schemas/11/users.sql | 10 ++--- synapse/storage/signatures.py | 8 ++-- synapse/storage/stream.py | 6 --- synapse/util/retryutils.py | 2 +- tests/federation/test_federation.py | 10 +++-- tests/handlers/test_federation.py | 9 +++++ tests/handlers/test_presence.py | 7 +++- tests/handlers/test_typing.py | 7 +++- tests/rest/client/v1/test_events.py | 9 ----- tests/storage/test_appservice.py | 12 ++---- tests/storage/test_base.py | 20 +++++----- tests/storage/test_registration.py | 36 +++++++++++------- tests/storage/test_roommember.py | 14 +++---- tests/utils.py | 11 +++++- 36 files changed, 296 insertions(+), 160 deletions(-) create mode 100644 synapse/storage/engines/__init__.py create mode 100644 synapse/storage/engines/maria.py create mode 100644 synapse/storage/engines/sqlite3.py (limited to 'tests') diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py index beab6ffc7a..b185b2f569 100755 --- a/synapse/app/homeserver.py +++ b/synapse/app/homeserver.py @@ -20,6 +20,7 @@ sys.dont_write_bytecode = True from synapse.storage import ( prepare_database, prepare_sqlite3_database, UpgradeDatabaseException, ) +from synapse.storage.engines import create_engine from synapse.server import HomeServer @@ -376,7 +377,7 @@ def setup(config_options): if name in ["MySQLdb", "mysql.connector"]: db_config.setdefault("args", {}).update({ "sql_mode": "TRADITIONAL", - "charset": "utf8", + "charset": "utf8mb4", "use_unicode": True, }) elif name == "sqlite3": @@ -388,6 +389,8 @@ def setup(config_options): else: raise RuntimeError("Unsupported database type '%s'" % (name,)) + database_engine = create_engine(name) + hs = SynapseHomeServer( config.server_name, domain_with_port=domain_with_port, @@ -398,6 +401,7 @@ def setup(config_options): config=config, content_addr=config.content_addr, version_string=version_string, + database_engine=database_engine, ) hs.create_resource_tree( @@ -409,12 +413,14 @@ def setup(config_options): logger.info("Preparing database: %s...", db_name) try: - # with sqlite3.connect(db_name) as db_conn: - # prepare_sqlite3_database(db_conn) - # prepare_database(db_conn) - import mysql.connector - db_conn = mysql.connector.connect(**db_config.get("args", {})) - prepare_database(db_conn) + db_conn = database_engine.module.connect(**db_config.get("args", {})) + + if name == "sqlite3": + prepare_sqlite3_database(db_conn) + + prepare_database(db_conn, database_engine) + + db_conn.commit() except UpgradeDatabaseException: sys.stderr.write( "\nFailed to upgrade database.\n" diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index abde7d0df5..f8053484cf 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -77,9 +77,6 @@ class DataStore(RoomMemberStore, RoomStore, self.min_token_deferred = self._get_min_token() self.min_token = None - self._next_stream_id_lock = threading.Lock() - self._next_stream_id = int(hs.get_clock().time_msec()) * 1000 - def insert_client_ip(self, user, access_token, device_id, ip, user_agent): return self._simple_upsert( "user_ips", @@ -127,19 +124,21 @@ class UpgradeDatabaseException(PrepareDatabaseException): pass -def prepare_database(db_conn): +def prepare_database(db_conn, database_engine): """Prepares a database for usage. Will either create all necessary tables or upgrade from an older schema version. """ try: cur = db_conn.cursor() - version_info = _get_or_create_schema_state(cur) + version_info = _get_or_create_schema_state(cur, database_engine) if version_info: user_version, delta_files, upgraded = version_info - _upgrade_existing_database(cur, user_version, delta_files, upgraded) + _upgrade_existing_database( + cur, user_version, delta_files, upgraded, database_engine + ) else: - _setup_new_database(cur) + _setup_new_database(cur, database_engine) # cur.execute("PRAGMA user_version = %d" % (SCHEMA_VERSION,)) @@ -150,7 +149,7 @@ def prepare_database(db_conn): raise -def _setup_new_database(cur): +def _setup_new_database(cur, database_engine): """Sets up the database by finding a base set of "full schemas" and then applying any necessary deltas. @@ -210,7 +209,7 @@ def _setup_new_database(cur): executescript(cur, sql_loc) cur.execute( - _convert_param_style( + database_engine.convert_param_style( "REPLACE INTO schema_version (version, upgraded)" " VALUES (?,?)" ), @@ -221,12 +220,13 @@ def _setup_new_database(cur): cur, current_version=max_current_ver, applied_delta_files=[], - upgraded=False + upgraded=False, + database_engine=database_engine, ) def _upgrade_existing_database(cur, current_version, applied_delta_files, - upgraded): + upgraded, database_engine): """Upgrades an existing database. Delta files can either be SQL stored in *.sql files, or python modules @@ -335,26 +335,22 @@ def _upgrade_existing_database(cur, current_version, applied_delta_files, # Mark as done. cur.execute( - _convert_param_style( + database_engine.convert_param_style( "INSERT INTO applied_schema_deltas (version, file)" - " VALUES (?,?)" + " VALUES (?,?)", ), (v, relative_path) ) cur.execute( - _convert_param_style( + database_engine.convert_param_style( "REPLACE INTO schema_version (version, upgraded)" - " VALUES (?,?)" + " VALUES (?,?)", ), (v, True) ) -def _convert_param_style(sql): - return sql.replace("?", "%s") - - def get_statements(f): statement_buffer = "" in_comment = False # If we're in a /* ... */ style comment @@ -409,7 +405,7 @@ def executescript(txn, schema_path): txn.execute(statement) -def _get_or_create_schema_state(txn): +def _get_or_create_schema_state(txn, database_engine): try: # Bluntly try creating the schema_version tables. schema_path = os.path.join( @@ -426,7 +422,7 @@ def _get_or_create_schema_state(txn): if current_version: txn.execute( - _convert_param_style( + database_engine.convert_param_style( "SELECT file FROM applied_schema_deltas WHERE version >= ?" ), (current_version,) @@ -446,6 +442,8 @@ def prepare_sqlite3_database(db_conn): new. This only affects sqlite databases since they were the only ones supported at the time. """ + import sqlite3 + with db_conn: schema_path = os.path.join( dir_path, "schema", "schema_version.sql", @@ -466,7 +464,8 @@ def prepare_sqlite3_database(db_conn): db_conn.execute( _convert_param_style( "REPLACE INTO schema_version (version, upgraded)" - " VALUES (?,?)" + " VALUES (?,?)", + sqlite3 ), (row[0], False) ) diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 76ec3ee93f..047d100f46 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -29,6 +29,7 @@ import functools import simplejson as json import sys import time +import threading logger = logging.getLogger(__name__) @@ -118,19 +119,16 @@ def cached(max_entries=1000, num_args=1): return wrap -def _convert_param_style(sql): - return sql.replace("?", "%s") - - class LoggingTransaction(object): """An object that almost-transparently proxies for the 'txn' object passed to the constructor. Adds logging and metrics to the .execute() method.""" - __slots__ = ["txn", "name"] + __slots__ = ["txn", "name", "database_engine"] - def __init__(self, txn, name): + def __init__(self, txn, name, database_engine): object.__setattr__(self, "txn", txn) object.__setattr__(self, "name", name) + object.__setattr__(self, "database_engine", database_engine) def __getattr__(self, name): return getattr(self.txn, name) @@ -142,7 +140,7 @@ class LoggingTransaction(object): # TODO(paul): Maybe use 'info' and 'debug' for values? sql_logger.debug("[SQL] {%s} %s", self.name, sql) - sql = _convert_param_style(sql) + sql = self.database_engine.convert_param_style(sql) try: if args and args[0]: @@ -227,9 +225,14 @@ class SQLBaseStore(object): self._get_event_cache = LruCache(hs.config.event_cache_size) + self.database_engine = hs.database_engine + # Pretend the getEventCache is just another named cache caches_by_name["*getEvent*"] = self._get_event_cache + self._next_stream_id_lock = threading.Lock() + self._next_stream_id = int(hs.get_clock().time_msec()) * 1000 + def start_profiling(self): self._previous_loop_ts = self._clock.time_msec() @@ -281,7 +284,10 @@ class SQLBaseStore(object): sql_scheduling_timer.inc_by(time.time() * 1000 - start_time) transaction_logger.debug("[TXN START] {%s}", name) try: - return func(LoggingTransaction(txn, name), *args, **kwargs) + return func( + LoggingTransaction(txn, name, self.database_engine), + *args, **kwargs + ) except: logger.exception("[TXN FAIL] {%s}", name) raise @@ -588,7 +594,7 @@ class SQLBaseStore(object): select_sql = "SELECT %s FROM %s WHERE %s" % ( ", ".join(retcols), table, - " AND ".join("%s = ?" % (k) for k in keyvalues) + " AND ".join("%s = ?" % (k,) for k in keyvalues) ) txn.execute(select_sql, keyvalues.values()) @@ -836,6 +842,12 @@ class SQLBaseStore(object): result = txn.fetchone() return result[0] if result else None + def get_next_stream_id(self): + with self._next_stream_id_lock: + i = self._next_stream_id + self._next_stream_id += 1 + return i + class _RollbackButIsFineException(Exception): """ This exception is used to rollback a transaction without implying diff --git a/synapse/storage/engines/__init__.py b/synapse/storage/engines/__init__.py new file mode 100644 index 0000000000..709b6f88ac --- /dev/null +++ b/synapse/storage/engines/__init__.py @@ -0,0 +1,35 @@ +# -*- coding: utf-8 -*- +# Copyright 2015 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from .maria import MariaEngine +from .sqlite3 import Sqlite3Engine + + +SUPPORTED_MODULE = { + "sqlite3": Sqlite3Engine, + "mysql.connector": MariaEngine, +} + + +def create_engine(name): + engine_class = SUPPORTED_MODULE.get(name, None) + + if engine_class: + module = __import__(name) + return engine_class(module) + + raise RuntimeError( + "Unsupported database engine '%s'" % (name,) + ) diff --git a/synapse/storage/engines/maria.py b/synapse/storage/engines/maria.py new file mode 100644 index 0000000000..df47763647 --- /dev/null +++ b/synapse/storage/engines/maria.py @@ -0,0 +1,30 @@ +# -*- coding: utf-8 -*- +# Copyright 2015 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +import types + + +class MariaEngine(object): + def __init__(self, database_module): + self.module = database_module + + def convert_param_style(self, sql): + return sql.replace("?", "%s") + + def encode_parameter(self, param): + if isinstance(param, types.BufferType): + return str(param) + return param diff --git a/synapse/storage/engines/sqlite3.py b/synapse/storage/engines/sqlite3.py new file mode 100644 index 0000000000..639cdea41d --- /dev/null +++ b/synapse/storage/engines/sqlite3.py @@ -0,0 +1,25 @@ +# -*- coding: utf-8 -*- +# Copyright 2015 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +class Sqlite3Engine(object): + def __init__(self, database_module): + self.module = database_module + + def convert_param_style(self, sql): + return sql + + def encode_parameter(self, param): + return param diff --git a/synapse/storage/keys.py b/synapse/storage/keys.py index e6975a945b..25fef79434 100644 --- a/synapse/storage/keys.py +++ b/synapse/storage/keys.py @@ -64,7 +64,7 @@ class KeyStore(SQLBaseStore): "fingerprint": fingerprint, "from_server": from_server, "ts_added_ms": time_now_ms, - "tls_certificate": tls_certificate_bytes, + "tls_certificate": buffer(tls_certificate_bytes), }, ) @@ -113,6 +113,6 @@ class KeyStore(SQLBaseStore): "key_id": "%s:%s" % (verify_key.alg, verify_key.version), "from_server": from_server, "ts_added_ms": time_now_ms, - "verify_key": verify_key.encode(), + "verify_key": buffer(verify_key.encode()), }, ) diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py index 7258f7b2a5..0c785ec989 100644 --- a/synapse/storage/registration.py +++ b/synapse/storage/registration.py @@ -42,6 +42,7 @@ class RegistrationStore(SQLBaseStore): yield self._simple_insert( "access_tokens", { + "id": self.get_next_stream_id(), "user_id": user_id, "token": token }, @@ -78,8 +79,11 @@ class RegistrationStore(SQLBaseStore): # it's possible for this to get a conflict, but only for a single user # since tokens are namespaced based on their user ID - txn.execute("INSERT INTO access_tokens(user_id, token) " + - "VALUES (?,?)", [user_id, token]) + txn.execute( + "INSERT INTO access_tokens(id, user_id, token)" + " VALUES (?,?,?)", + (self.get_next_stream_id(), user_id, token,) + ) @defer.inlineCallbacks def get_user_by_id(self, user_id): diff --git a/synapse/storage/schema/delta/12/v12.sql b/synapse/storage/schema/delta/12/v12.sql index b526109e6e..90ac474859 100644 --- a/synapse/storage/schema/delta/12/v12.sql +++ b/synapse/storage/schema/delta/12/v12.sql @@ -18,7 +18,7 @@ CREATE TABLE IF NOT EXISTS rejections( reason VARCHAR(255) NOT NULL, last_check VARCHAR(255) NOT NULL, UNIQUE (event_id) -) ENGINE = INNODB; +) ; -- Push notification endpoints that users have configured CREATE TABLE IF NOT EXISTS pushers ( @@ -37,7 +37,7 @@ CREATE TABLE IF NOT EXISTS pushers ( last_success BIGINT, failing_since BIGINT, UNIQUE (app_id, pushkey) -) ENGINE = INNODB; +) ; CREATE TABLE IF NOT EXISTS push_rules ( id BIGINT PRIMARY KEY, @@ -48,7 +48,7 @@ CREATE TABLE IF NOT EXISTS push_rules ( conditions VARCHAR(255) NOT NULL, actions VARCHAR(255) NOT NULL, UNIQUE(user_name, rule_id) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS push_rules_user_name on push_rules (user_name); @@ -56,7 +56,7 @@ CREATE TABLE IF NOT EXISTS user_filters( user_id VARCHAR(255), filter_id BIGINT, filter_json BLOB -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS user_filters_by_user_id_filter_id ON user_filters( user_id, filter_id diff --git a/synapse/storage/schema/delta/13/v13.sql b/synapse/storage/schema/delta/13/v13.sql index f0a5daf445..4953b6323c 100644 --- a/synapse/storage/schema/delta/13/v13.sql +++ b/synapse/storage/schema/delta/13/v13.sql @@ -20,7 +20,7 @@ CREATE TABLE IF NOT EXISTS application_services( hs_token VARCHAR(255), sender VARCHAR(255), UNIQUE(token) -) ENGINE = INNODB; +) ; CREATE TABLE IF NOT EXISTS application_services_regex( id BIGINT PRIMARY KEY, @@ -28,4 +28,4 @@ CREATE TABLE IF NOT EXISTS application_services_regex( namespace INTEGER, /* enum[room_id|room_alias|user_id] */ regex VARCHAR(255), FOREIGN KEY(as_id) REFERENCES application_services(id) -) ENGINE = INNODB; +) ; diff --git a/synapse/storage/schema/delta/14/v14.sql b/synapse/storage/schema/delta/14/v14.sql index a1260c5c1f..3bda073c94 100644 --- a/synapse/storage/schema/delta/14/v14.sql +++ b/synapse/storage/schema/delta/14/v14.sql @@ -4,6 +4,6 @@ CREATE TABLE IF NOT EXISTS push_rules_enable ( rule_id VARCHAR(255) NOT NULL, enabled TINYINT, UNIQUE(user_name, rule_id) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS push_rules_enable_user_name on push_rules_enable (user_name); diff --git a/synapse/storage/schema/full_schemas/11/event_edges.sql b/synapse/storage/schema/full_schemas/11/event_edges.sql index 0f53488e92..336cd563df 100644 --- a/synapse/storage/schema/full_schemas/11/event_edges.sql +++ b/synapse/storage/schema/full_schemas/11/event_edges.sql @@ -17,7 +17,7 @@ CREATE TABLE IF NOT EXISTS event_forward_extremities( event_id VARCHAR(255) NOT NULL, room_id VARCHAR(255) NOT NULL, UNIQUE (event_id, room_id) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS ev_extrem_room ON event_forward_extremities(room_id); CREATE INDEX IF NOT EXISTS ev_extrem_id ON event_forward_extremities(event_id); @@ -27,7 +27,7 @@ CREATE TABLE IF NOT EXISTS event_backward_extremities( event_id VARCHAR(255) NOT NULL, room_id VARCHAR(255) NOT NULL, UNIQUE (event_id, room_id) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS ev_b_extrem_room ON event_backward_extremities(room_id); CREATE INDEX IF NOT EXISTS ev_b_extrem_id ON event_backward_extremities(event_id); @@ -39,7 +39,7 @@ CREATE TABLE IF NOT EXISTS event_edges( room_id VARCHAR(255) NOT NULL, is_state BOOL NOT NULL, UNIQUE (event_id, prev_event_id, room_id, is_state) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS ev_edges_id ON event_edges(event_id); CREATE INDEX IF NOT EXISTS ev_edges_prev_id ON event_edges(prev_event_id); @@ -49,7 +49,7 @@ CREATE TABLE IF NOT EXISTS room_depth( room_id VARCHAR(255) NOT NULL, min_depth INTEGER NOT NULL, UNIQUE (room_id) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS room_depth_room ON room_depth(room_id); @@ -59,7 +59,7 @@ create TABLE IF NOT EXISTS event_destinations( destination VARCHAR(255) NOT NULL, delivered_ts BIGINT DEFAULT 0, -- or 0 if not delivered UNIQUE (event_id, destination) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS event_destinations_id ON event_destinations(event_id); @@ -70,7 +70,7 @@ CREATE TABLE IF NOT EXISTS state_forward_extremities( type VARCHAR(255) NOT NULL, state_key VARCHAR(255) NOT NULL, UNIQUE (event_id, room_id) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS st_extrem_keys ON state_forward_extremities( room_id, type, state_key @@ -83,7 +83,7 @@ CREATE TABLE IF NOT EXISTS event_auth( auth_id VARCHAR(255) NOT NULL, room_id VARCHAR(255) NOT NULL, UNIQUE (event_id, auth_id, room_id) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS evauth_edges_id ON event_auth(event_id); CREATE INDEX IF NOT EXISTS evauth_edges_auth_id ON event_auth(auth_id); diff --git a/synapse/storage/schema/full_schemas/11/event_signatures.sql b/synapse/storage/schema/full_schemas/11/event_signatures.sql index 334d7c8680..11e611598b 100644 --- a/synapse/storage/schema/full_schemas/11/event_signatures.sql +++ b/synapse/storage/schema/full_schemas/11/event_signatures.sql @@ -18,7 +18,7 @@ CREATE TABLE IF NOT EXISTS event_content_hashes ( algorithm VARCHAR(255), hash BLOB, UNIQUE (event_id, algorithm) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS event_content_hashes_id ON event_content_hashes(event_id); @@ -28,7 +28,7 @@ CREATE TABLE IF NOT EXISTS event_reference_hashes ( algorithm VARCHAR(255), hash BLOB, UNIQUE (event_id, algorithm) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS event_reference_hashes_id ON event_reference_hashes(event_id); @@ -39,7 +39,7 @@ CREATE TABLE IF NOT EXISTS event_signatures ( key_id VARCHAR(255), signature BLOB, UNIQUE (event_id, signature_name, key_id) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS event_signatures_id ON event_signatures(event_id); @@ -50,6 +50,6 @@ CREATE TABLE IF NOT EXISTS event_edge_hashes( algorithm VARCHAR(255), hash BLOB, UNIQUE (event_id, prev_event_id, algorithm) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS event_edge_hashes_id ON event_edge_hashes(event_id); diff --git a/synapse/storage/schema/full_schemas/11/im.sql b/synapse/storage/schema/full_schemas/11/im.sql index 9849e969be..a0fb337629 100644 --- a/synapse/storage/schema/full_schemas/11/im.sql +++ b/synapse/storage/schema/full_schemas/11/im.sql @@ -25,7 +25,7 @@ CREATE TABLE IF NOT EXISTS events( outlier BOOL NOT NULL, depth BIGINT DEFAULT 0 NOT NULL, UNIQUE (event_id) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS events_stream_ordering ON events (stream_ordering); CREATE INDEX IF NOT EXISTS events_topological_ordering ON events (topological_ordering); @@ -38,7 +38,7 @@ CREATE TABLE IF NOT EXISTS event_json( internal_metadata BLOB NOT NULL, json BLOB NOT NULL, UNIQUE (event_id) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS event_json_room_id ON event_json(room_id); @@ -50,7 +50,7 @@ CREATE TABLE IF NOT EXISTS state_events( state_key VARCHAR(255) NOT NULL, prev_state VARCHAR(255), UNIQUE (event_id) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS state_events_room_id ON state_events (room_id); CREATE INDEX IF NOT EXISTS state_events_type ON state_events (type); @@ -64,7 +64,7 @@ CREATE TABLE IF NOT EXISTS current_state_events( state_key VARCHAR(255) NOT NULL, UNIQUE (event_id), UNIQUE (room_id, type, state_key) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS current_state_events_room_id ON current_state_events (room_id); CREATE INDEX IF NOT EXISTS current_state_events_type ON current_state_events (type); @@ -77,7 +77,7 @@ CREATE TABLE IF NOT EXISTS room_memberships( room_id VARCHAR(255) NOT NULL, membership VARCHAR(255) NOT NULL, UNIQUE (event_id) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS room_memberships_room_id ON room_memberships (room_id); CREATE INDEX IF NOT EXISTS room_memberships_user_id ON room_memberships (user_id); @@ -89,14 +89,14 @@ CREATE TABLE IF NOT EXISTS feedback( sender VARCHAR(255), room_id VARCHAR(255), UNIQUE (event_id) -) ENGINE = INNODB; +) ; CREATE TABLE IF NOT EXISTS topics( event_id VARCHAR(255) NOT NULL, room_id VARCHAR(255) NOT NULL, topic VARCHAR(255) NOT NULL, UNIQUE (event_id) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS topics_room_id ON topics(room_id); @@ -113,12 +113,12 @@ CREATE TABLE IF NOT EXISTS rooms( room_id VARCHAR(255) PRIMARY KEY NOT NULL, is_public BOOL, creator VARCHAR(255) -) ENGINE = INNODB; +) ; CREATE TABLE IF NOT EXISTS room_hosts( room_id VARCHAR(255) NOT NULL, host VARCHAR(255) NOT NULL, UNIQUE (room_id, host) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS room_hosts_room_id ON room_hosts (room_id); diff --git a/synapse/storage/schema/full_schemas/11/keys.sql b/synapse/storage/schema/full_schemas/11/keys.sql index c0f2ec29bb..a785cdb4c5 100644 --- a/synapse/storage/schema/full_schemas/11/keys.sql +++ b/synapse/storage/schema/full_schemas/11/keys.sql @@ -19,7 +19,7 @@ CREATE TABLE IF NOT EXISTS server_tls_certificates( ts_added_ms BIGINT, -- When the certifcate was added. tls_certificate BLOB, -- DER encoded x509 certificate. UNIQUE (server_name, fingerprint) -) ENGINE = INNODB; +) ; CREATE TABLE IF NOT EXISTS server_signature_keys( server_name VARCHAR(255), -- Server name. @@ -28,4 +28,4 @@ CREATE TABLE IF NOT EXISTS server_signature_keys( ts_added_ms BIGINT, -- When the key was added. verify_key BLOB, -- NACL verification key. UNIQUE (server_name, key_id) -) ENGINE = INNODB; +) ; diff --git a/synapse/storage/schema/full_schemas/11/media_repository.sql b/synapse/storage/schema/full_schemas/11/media_repository.sql index d9559f5902..27fe297af6 100644 --- a/synapse/storage/schema/full_schemas/11/media_repository.sql +++ b/synapse/storage/schema/full_schemas/11/media_repository.sql @@ -21,7 +21,7 @@ CREATE TABLE IF NOT EXISTS local_media_repository ( upload_name VARCHAR(255), -- The name the media was uploaded with. user_id VARCHAR(255), -- The user who uploaded the file. UNIQUE (media_id) -) ENGINE = INNODB; +) ; CREATE TABLE IF NOT EXISTS local_media_repository_thumbnails ( media_id VARCHAR(255), -- The id used to refer to the media. @@ -33,7 +33,7 @@ CREATE TABLE IF NOT EXISTS local_media_repository_thumbnails ( UNIQUE ( media_id, thumbnail_width, thumbnail_height, thumbnail_type ) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS local_media_repository_thumbnails_media_id ON local_media_repository_thumbnails (media_id); @@ -47,7 +47,7 @@ CREATE TABLE IF NOT EXISTS remote_media_cache ( media_length INTEGER, -- Length of the media in bytes. filesystem_id VARCHAR(255), -- The name used to store the media on disk. UNIQUE (media_origin, media_id) -) ENGINE = INNODB; +) ; CREATE TABLE IF NOT EXISTS remote_media_cache_thumbnails ( media_origin VARCHAR(255), -- The remote HS the media came from. @@ -62,7 +62,7 @@ CREATE TABLE IF NOT EXISTS remote_media_cache_thumbnails ( media_origin, media_id, thumbnail_width, thumbnail_height, thumbnail_type ) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS remote_media_cache_thumbnails_media_id ON remote_media_cache_thumbnails (media_id); diff --git a/synapse/storage/schema/full_schemas/11/presence.sql b/synapse/storage/schema/full_schemas/11/presence.sql index 8031321083..b48b110ae9 100644 --- a/synapse/storage/schema/full_schemas/11/presence.sql +++ b/synapse/storage/schema/full_schemas/11/presence.sql @@ -18,7 +18,7 @@ CREATE TABLE IF NOT EXISTS presence( status_msg VARCHAR(255), mtime BIGINT, -- miliseconds since last state change UNIQUE(user_id) -) ENGINE = INNODB; +) ; -- For each of /my/ users which possibly-remote users are allowed to see their -- presence state @@ -26,7 +26,7 @@ CREATE TABLE IF NOT EXISTS presence_allow_inbound( observed_user_id VARCHAR(255) NOT NULL, observer_user_id VARCHAR(255), -- a UserID, UNIQUE(observed_user_id) -) ENGINE = INNODB; +) ; -- For each of /my/ users (watcher), which possibly-remote users are they -- watching? @@ -35,4 +35,4 @@ CREATE TABLE IF NOT EXISTS presence_list( observed_user_id VARCHAR(255), -- a UserID, accepted BOOLEAN, UNIQUE(user_id) -) ENGINE = INNODB; +) ; diff --git a/synapse/storage/schema/full_schemas/11/profiles.sql b/synapse/storage/schema/full_schemas/11/profiles.sql index 552645c56f..92da48f97e 100644 --- a/synapse/storage/schema/full_schemas/11/profiles.sql +++ b/synapse/storage/schema/full_schemas/11/profiles.sql @@ -14,7 +14,7 @@ */ CREATE TABLE IF NOT EXISTS profiles( user_id VARCHAR(255) NOT NULL, - displayname VARBINARY(255), + displayname VARCHAR(255), avatar_url VARCHAR(255), UNIQUE(user_id) -) ENGINE = INNODB; +) ; diff --git a/synapse/storage/schema/full_schemas/11/redactions.sql b/synapse/storage/schema/full_schemas/11/redactions.sql index ba93e860f6..9b52a2012a 100644 --- a/synapse/storage/schema/full_schemas/11/redactions.sql +++ b/synapse/storage/schema/full_schemas/11/redactions.sql @@ -16,7 +16,7 @@ CREATE TABLE IF NOT EXISTS redactions ( event_id VARCHAR(255) NOT NULL, redacts VARCHAR(255) NOT NULL, UNIQUE (event_id) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS redactions_event_id ON redactions (event_id); CREATE INDEX IF NOT EXISTS redactions_redacts ON redactions (redacts); diff --git a/synapse/storage/schema/full_schemas/11/room_aliases.sql b/synapse/storage/schema/full_schemas/11/room_aliases.sql index 1e706aac2b..220df87573 100644 --- a/synapse/storage/schema/full_schemas/11/room_aliases.sql +++ b/synapse/storage/schema/full_schemas/11/room_aliases.sql @@ -14,12 +14,12 @@ */ CREATE TABLE IF NOT EXISTS room_aliases( - room_alias VARCHAR(255) NOT NULL, + room_alias VARBINARY(255) NOT NULL, room_id VARCHAR(255) NOT NULL, UNIQUE (room_alias) -) ENGINE = INNODB; +) ; CREATE TABLE IF NOT EXISTS room_alias_servers( - room_alias VARCHAR(255) NOT NULL, + room_alias VARBINARY(255) NOT NULL, server VARCHAR(255) NOT NULL -) ENGINE = INNODB; +) ; diff --git a/synapse/storage/schema/full_schemas/11/state.sql b/synapse/storage/schema/full_schemas/11/state.sql index be9dc2920d..40584a325f 100644 --- a/synapse/storage/schema/full_schemas/11/state.sql +++ b/synapse/storage/schema/full_schemas/11/state.sql @@ -17,7 +17,7 @@ CREATE TABLE IF NOT EXISTS state_groups( id VARCHAR(20) PRIMARY KEY, room_id VARCHAR(255) NOT NULL, event_id VARCHAR(255) NOT NULL -) ENGINE = INNODB; +) ; CREATE TABLE IF NOT EXISTS state_groups_state( state_group VARCHAR(20) NOT NULL, @@ -25,13 +25,13 @@ CREATE TABLE IF NOT EXISTS state_groups_state( type VARCHAR(255) NOT NULL, state_key VARCHAR(255) NOT NULL, event_id VARCHAR(255) NOT NULL -) ENGINE = INNODB; +) ; CREATE TABLE IF NOT EXISTS event_to_state_groups( event_id VARCHAR(255) NOT NULL, state_group VARCHAR(255) NOT NULL, UNIQUE (event_id) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS state_groups_id ON state_groups(id); diff --git a/synapse/storage/schema/full_schemas/11/transactions.sql b/synapse/storage/schema/full_schemas/11/transactions.sql index bd13bba8c2..d33bdfb301 100644 --- a/synapse/storage/schema/full_schemas/11/transactions.sql +++ b/synapse/storage/schema/full_schemas/11/transactions.sql @@ -21,7 +21,7 @@ CREATE TABLE IF NOT EXISTS received_transactions( response_json BLOB, has_been_referenced BOOL default 0, -- Whether thishas been referenced by a prev_tx UNIQUE (transaction_id, origin) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS transactions_have_ref ON received_transactions(origin, has_been_referenced);-- WHERE has_been_referenced = 0; @@ -35,7 +35,7 @@ CREATE TABLE IF NOT EXISTS sent_transactions( response_code INTEGER DEFAULT 0, response_json BLOB, ts BIGINT -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS sent_transaction_dest ON sent_transactions(destination); CREATE INDEX IF NOT EXISTS sent_transaction_txn_id ON sent_transactions(transaction_id); @@ -51,7 +51,7 @@ CREATE TABLE IF NOT EXISTS transaction_id_to_pdu( pdu_id VARCHAR(255), pdu_origin VARCHAR(255), UNIQUE (transaction_id, destination) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS transaction_id_to_pdu_dest ON transaction_id_to_pdu(destination); @@ -60,4 +60,4 @@ CREATE TABLE IF NOT EXISTS destinations( destination VARCHAR(255) PRIMARY KEY, retry_last_ts BIGINT, retry_interval INTEGER -) ENGINE = INNODB; +) ; diff --git a/synapse/storage/schema/full_schemas/11/users.sql b/synapse/storage/schema/full_schemas/11/users.sql index 55bffb22f3..28909f5805 100644 --- a/synapse/storage/schema/full_schemas/11/users.sql +++ b/synapse/storage/schema/full_schemas/11/users.sql @@ -14,20 +14,20 @@ */ CREATE TABLE IF NOT EXISTS users( name VARCHAR(255), - password_hash VARBINARY(255), + password_hash VARCHAR(255), creation_ts BIGINT, admin BOOL DEFAULT 0 NOT NULL, UNIQUE(name) -) ENGINE = INNODB; +) ; CREATE TABLE IF NOT EXISTS access_tokens( - id INTEGER PRIMARY KEY AUTO_INCREMENT, + id BIGINT PRIMARY KEY, user_id VARCHAR(255) NOT NULL, device_id VARCHAR(255), token VARCHAR(255) NOT NULL, last_used BIGINT, UNIQUE(token) -) ENGINE = INNODB; +) ; CREATE TABLE IF NOT EXISTS user_ips ( user VARCHAR(255) NOT NULL, @@ -37,6 +37,6 @@ CREATE TABLE IF NOT EXISTS user_ips ( user_agent VARCHAR(255) NOT NULL, last_seen BIGINT NOT NULL, UNIQUE (user, access_token, ip, user_agent) -) ENGINE = INNODB; +) ; CREATE INDEX IF NOT EXISTS user_ips_user ON user_ips(user); diff --git a/synapse/storage/signatures.py b/synapse/storage/signatures.py index 35bba854f9..f051828630 100644 --- a/synapse/storage/signatures.py +++ b/synapse/storage/signatures.py @@ -54,7 +54,7 @@ class SignatureStore(SQLBaseStore): { "event_id": event_id, "algorithm": algorithm, - "hash": hash_bytes, + "hash": buffer(hash_bytes), }, ) @@ -116,7 +116,7 @@ class SignatureStore(SQLBaseStore): { "event_id": event_id, "algorithm": algorithm, - "hash": hash_bytes, + "hash": buffer(hash_bytes), }, ) @@ -160,7 +160,7 @@ class SignatureStore(SQLBaseStore): "event_id": event_id, "signature_name": signature_name, "key_id": key_id, - "signature": signature_bytes, + "signature": buffer(signature_bytes), }, ) @@ -193,6 +193,6 @@ class SignatureStore(SQLBaseStore): "event_id": event_id, "prev_event_id": prev_event_id, "algorithm": algorithm, - "hash": hash_bytes, + "hash": buffer(hash_bytes), }, ) diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index 3a310cd003..e6bb5a8077 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -433,12 +433,6 @@ class StreamStore(SQLBaseStore): defer.returnValue(self.min_token) - def get_next_stream_id(self): - with self._next_stream_id_lock: - i = self._next_stream_id - self._next_stream_id += 1 - return i - def _get_room_events_max_id_txn(self, txn): txn.execute( "SELECT MAX(stream_ordering) as m FROM events" diff --git a/synapse/util/retryutils.py b/synapse/util/retryutils.py index 4e82232796..a42138f556 100644 --- a/synapse/util/retryutils.py +++ b/synapse/util/retryutils.py @@ -60,7 +60,7 @@ def get_retry_limiter(destination, clock, store, **kwargs): if retry_timings: retry_last_ts, retry_interval = ( - retry_timings.retry_last_ts, retry_timings.retry_interval + retry_timings["retry_last_ts"], retry_timings["retry_interval"] ) now = int(clock.time_msec()) diff --git a/tests/federation/test_federation.py b/tests/federation/test_federation.py index 2ecd00d2ad..a4ef60b911 100644 --- a/tests/federation/test_federation.py +++ b/tests/federation/test_federation.py @@ -24,8 +24,6 @@ from ..utils import MockHttpResource, MockClock, setup_test_homeserver from synapse.federation import initialize_http_replication from synapse.events import FrozenEvent -from synapse.storage.transactions import DestinationsTable - def make_pdu(prev_pdus=[], **kwargs): """Provide some default fields for making a PduTuple.""" @@ -57,8 +55,14 @@ class FederationTestCase(unittest.TestCase): self.mock_persistence.get_received_txn_response.return_value = ( defer.succeed(None) ) + + retry_timings_res = { + "destination": "", + "retry_last_ts": 0, + "retry_interval": 0, + } self.mock_persistence.get_destination_retry_timings.return_value = ( - defer.succeed(DestinationsTable.EntryType("", 0, 0)) + defer.succeed(retry_timings_res) ) self.mock_persistence.get_auth_chain.return_value = [] self.clock = MockClock() diff --git a/tests/handlers/test_federation.py b/tests/handlers/test_federation.py index c13ade3286..08d2404b6c 100644 --- a/tests/handlers/test_federation.py +++ b/tests/handlers/test_federation.py @@ -87,6 +87,15 @@ class FederationTestCase(unittest.TestCase): self.datastore.get_room.return_value = defer.succeed(True) self.auth.check_host_in_room.return_value = defer.succeed(True) + retry_timings_res = { + "destination": "", + "retry_last_ts": 0, + "retry_interval": 0, + } + self.datastore.get_destination_retry_timings.return_value = ( + defer.succeed(retry_timings_res) + ) + def have_events(event_ids): return defer.succeed({}) self.datastore.have_events.side_effect = have_events diff --git a/tests/handlers/test_presence.py b/tests/handlers/test_presence.py index 04eba4289e..9b0e606918 100644 --- a/tests/handlers/test_presence.py +++ b/tests/handlers/test_presence.py @@ -194,8 +194,13 @@ class MockedDatastorePresenceTestCase(PresenceTestCase): return datastore def setUp_datastore_federation_mocks(self, datastore): + retry_timings_res = { + "destination": "", + "retry_last_ts": 0, + "retry_interval": 0, + } datastore.get_destination_retry_timings.return_value = ( - defer.succeed(DestinationsTable.EntryType("", 0, 0)) + defer.succeed(retry_timings_res) ) def get_received_txn_response(*args): diff --git a/tests/handlers/test_typing.py b/tests/handlers/test_typing.py index bf34b7ccbd..2d76b23564 100644 --- a/tests/handlers/test_typing.py +++ b/tests/handlers/test_typing.py @@ -96,8 +96,13 @@ class TypingNotificationsTestCase(unittest.TestCase): self.event_source = hs.get_event_sources().sources["typing"] self.datastore = hs.get_datastore() + retry_timings_res = { + "destination": "", + "retry_last_ts": 0, + "retry_interval": 0, + } self.datastore.get_destination_retry_timings.return_value = ( - defer.succeed(DestinationsTable.EntryType("", 0, 0)) + defer.succeed(retry_timings_res) ) def get_received_txn_response(*args): diff --git a/tests/rest/client/v1/test_events.py b/tests/rest/client/v1/test_events.py index 36b0f2ff6d..445272e323 100644 --- a/tests/rest/client/v1/test_events.py +++ b/tests/rest/client/v1/test_events.py @@ -115,12 +115,6 @@ class EventStreamPermissionsTestCase(RestTestCase): hs = yield setup_test_homeserver( http_client=None, replication_layer=Mock(), - clock=Mock(spec=[ - "call_later", - "cancel_call_later", - "time_msec", - "time" - ]), ratelimiter=NonCallableMock(spec_set=[ "send_message", ]), @@ -132,9 +126,6 @@ class EventStreamPermissionsTestCase(RestTestCase): hs.get_handlers().federation_handler = Mock() - hs.get_clock().time_msec.return_value = 1000000 - hs.get_clock().time.return_value = 1000 - synapse.rest.client.v1.register.register_servlets(hs, self.mock_resource) synapse.rest.client.v1.events.register_servlets(hs, self.mock_resource) synapse.rest.client.v1.room.register_servlets(hs, self.mock_resource) diff --git a/tests/storage/test_appservice.py b/tests/storage/test_appservice.py index ca5b92ec85..2ad55c8462 100644 --- a/tests/storage/test_appservice.py +++ b/tests/storage/test_appservice.py @@ -16,22 +16,18 @@ from tests import unittest from twisted.internet import defer from synapse.appservice import ApplicationService -from synapse.server import HomeServer from synapse.storage.appservice import ApplicationServiceStore -from mock import Mock -from tests.utils import SQLiteMemoryDbPool, MockClock +from tests.utils import setup_test_homeserver class ApplicationServiceStoreTestCase(unittest.TestCase): @defer.inlineCallbacks def setUp(self): - db_pool = SQLiteMemoryDbPool() - yield db_pool.prepare() - hs = HomeServer( - "test", db_pool=db_pool, clock=MockClock(), config=Mock() - ) + hs = yield setup_test_homeserver() + db_pool = hs.get_db_pool() + self.as_token = "token1" db_pool.runQuery( "INSERT INTO application_services(token) VALUES(?)", diff --git a/tests/storage/test_base.py b/tests/storage/test_base.py index 7f5845cf0c..5c17d30148 100644 --- a/tests/storage/test_base.py +++ b/tests/storage/test_base.py @@ -24,6 +24,7 @@ from collections import OrderedDict from synapse.server import HomeServer from synapse.storage._base import SQLBaseStore +from synapse.storage.engines import create_engine class SQLBaseStoreTestCase(unittest.TestCase): @@ -40,7 +41,12 @@ class SQLBaseStoreTestCase(unittest.TestCase): config = Mock() config.event_cache_size = 1 - hs = HomeServer("test", db_pool=self.db_pool, config=config) + hs = HomeServer( + "test", + db_pool=self.db_pool, + config=config, + database_engine=create_engine("sqlite3"), + ) self.datastore = SQLBaseStore(hs) @@ -86,8 +92,7 @@ class SQLBaseStoreTestCase(unittest.TestCase): self.assertEquals("Value", value) self.mock_txn.execute.assert_called_with( - "SELECT retcol FROM tablename WHERE keycol = ? " - "ORDER BY rowid asc", + "SELECT retcol FROM tablename WHERE keycol = ?", ["TheKey"] ) @@ -104,8 +109,7 @@ class SQLBaseStoreTestCase(unittest.TestCase): self.assertEquals({"colA": 1, "colB": 2, "colC": 3}, ret) self.mock_txn.execute.assert_called_with( - "SELECT colA, colB, colC FROM tablename WHERE keycol = ? " - "ORDER BY rowid asc", + "SELECT colA, colB, colC FROM tablename WHERE keycol = ?", ["TheKey"] ) @@ -139,8 +143,7 @@ class SQLBaseStoreTestCase(unittest.TestCase): self.assertEquals([{"colA": 1}, {"colA": 2}, {"colA": 3}], ret) self.mock_txn.execute.assert_called_with( - "SELECT colA FROM tablename WHERE keycol = ? " - "ORDER BY rowid asc", + "SELECT colA FROM tablename WHERE keycol = ?", ["A set"] ) @@ -189,8 +192,7 @@ class SQLBaseStoreTestCase(unittest.TestCase): self.assertEquals({"columname": "Old Value"}, ret) self.mock_txn.execute.assert_has_calls([ - call('SELECT columname FROM tablename WHERE keycol = ? ' - 'ORDER BY rowid asc', + call('SELECT columname FROM tablename WHERE keycol = ?', ['TheKey']), call("UPDATE tablename SET columname = ? WHERE keycol = ?", ["New Value", "TheKey"]) diff --git a/tests/storage/test_registration.py b/tests/storage/test_registration.py index e0b81f2b57..78f6004204 100644 --- a/tests/storage/test_registration.py +++ b/tests/storage/test_registration.py @@ -42,28 +42,38 @@ class RegistrationStoreTestCase(unittest.TestCase): self.assertEquals( # TODO(paul): Surely this field should be 'user_id', not 'name' # Additionally surely it shouldn't come in a 1-element list - [{"name": self.user_id, "password_hash": self.pwhash}], + {"name": self.user_id, "password_hash": self.pwhash}, (yield self.store.get_user_by_id(self.user_id)) ) - self.assertEquals( - {"admin": 0, - "device_id": None, - "name": self.user_id, - "token_id": 1}, - (yield self.store.get_user_by_token(self.tokens[0])) + result = yield self.store.get_user_by_token(self.tokens[1]) + + self.assertDictContainsSubset( + { + "admin": 0, + "device_id": None, + "name": self.user_id, + }, + result ) + self.assertTrue("token_id" in result) + @defer.inlineCallbacks def test_add_tokens(self): yield self.store.register(self.user_id, self.tokens[0], self.pwhash) yield self.store.add_access_token_to_user(self.user_id, self.tokens[1]) - self.assertEquals( - {"admin": 0, - "device_id": None, - "name": self.user_id, - "token_id": 2}, - (yield self.store.get_user_by_token(self.tokens[1])) + result = yield self.store.get_user_by_token(self.tokens[1]) + + self.assertDictContainsSubset( + { + "admin": 0, + "device_id": None, + "name": self.user_id, + }, + result ) + self.assertTrue("token_id" in result) + diff --git a/tests/storage/test_roommember.py b/tests/storage/test_roommember.py index 811fea544b..785953cc89 100644 --- a/tests/storage/test_roommember.py +++ b/tests/storage/test_roommember.py @@ -119,7 +119,7 @@ class RoomMemberStoreTestCase(unittest.TestCase): yield self.inject_room_member(self.room, self.u_alice, Membership.JOIN) self.assertEquals( - ["test"], + {"test"}, (yield self.store.get_joined_hosts_for_room(self.room.to_string())) ) @@ -127,7 +127,7 @@ class RoomMemberStoreTestCase(unittest.TestCase): yield self.inject_room_member(self.room, self.u_bob, Membership.JOIN) self.assertEquals( - ["test"], + {"test"}, (yield self.store.get_joined_hosts_for_room(self.room.to_string())) ) @@ -136,9 +136,9 @@ class RoomMemberStoreTestCase(unittest.TestCase): self.assertEquals( {"test", "elsewhere"}, - set((yield + (yield self.store.get_joined_hosts_for_room(self.room.to_string()) - )) + ) ) # Should still have both hosts @@ -146,15 +146,15 @@ class RoomMemberStoreTestCase(unittest.TestCase): self.assertEquals( {"test", "elsewhere"}, - set((yield + (yield self.store.get_joined_hosts_for_room(self.room.to_string()) - )) + ) ) # Should have only one host after other leaves yield self.inject_room_member(self.room, self.u_charlie, Membership.LEAVE) self.assertEquals( - ["test"], + {"test"}, (yield self.store.get_joined_hosts_for_room(self.room.to_string())) ) diff --git a/tests/utils.py b/tests/utils.py index 81e82a80df..cc038fecf1 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -17,6 +17,7 @@ from synapse.http.server import HttpServer from synapse.api.errors import cs_error, CodeMessageException, StoreError from synapse.api.constants import EventTypes from synapse.storage import prepare_database +from synapse.storage.engines import create_engine from synapse.server import HomeServer from synapse.util.logcontext import LoggingContext @@ -44,18 +45,23 @@ def setup_test_homeserver(name="test", datastore=None, config=None, **kargs): config.event_cache_size = 1 config.disable_registration = False + if "clock" not in kargs: + kargs["clock"] = MockClock() + if datastore is None: db_pool = SQLiteMemoryDbPool() yield db_pool.prepare() hs = HomeServer( name, db_pool=db_pool, config=config, version_string="Synapse/tests", + database_engine=create_engine("sqlite3"), **kargs ) else: hs = HomeServer( name, db_pool=None, datastore=datastore, config=config, version_string="Synapse/tests", + database_engine=create_engine("sqlite3"), **kargs ) @@ -227,7 +233,10 @@ class SQLiteMemoryDbPool(ConnectionPool, object): ) def prepare(self): - return self.runWithConnection(prepare_database) + engine = create_engine("sqlite3") + return self.runWithConnection( + lambda conn: prepare_database(conn, engine) + ) class MemoryDataStore(object): -- cgit 1.5.1 From 22d7a593062eb06081d0cd1dc71e4fc79fc1cb85 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 8 Apr 2015 16:57:14 +0100 Subject: Fix tests after commit 9a0579 --- tests/storage/test_base.py | 6 ++++++ 1 file changed, 6 insertions(+) (limited to 'tests') diff --git a/tests/storage/test_base.py b/tests/storage/test_base.py index 5c17d30148..a64d2b821e 100644 --- a/tests/storage/test_base.py +++ b/tests/storage/test_base.py @@ -33,12 +33,18 @@ class SQLBaseStoreTestCase(unittest.TestCase): def setUp(self): self.db_pool = Mock(spec=["runInteraction"]) self.mock_txn = Mock() + self.mock_conn = Mock(spec_set=["cursor"]) + self.mock_conn.cursor.return_value = self.mock_txn # Our fake runInteraction just runs synchronously inline def runInteraction(func, *args, **kwargs): return defer.succeed(func(self.mock_txn, *args, **kwargs)) self.db_pool.runInteraction = runInteraction + def runWithConnection(func, *args, **kwargs): + return defer.succeed(func(self.mock_conn, *args, **kwargs)) + self.db_pool.runWithConnection = runWithConnection + config = Mock() config.event_cache_size = 1 hs = HomeServer( -- cgit 1.5.1 From 0268d40281313c9a89e7b4356ae2e5f77a622857 Mon Sep 17 00:00:00 2001 From: "Paul \"LeoNerd\" Evans" Date: Wed, 15 Apr 2015 23:09:35 +0100 Subject: Have TypingNotificationEventSource.get_new_events_for_user() return a deferred, for consistency and extensibility --- synapse/handlers/typing.py | 2 +- tests/handlers/test_typing.py | 18 ++++++++++++------ tests/rest/client/v1/test_typing.py | 3 ++- 3 files changed, 15 insertions(+), 8 deletions(-) (limited to 'tests') diff --git a/synapse/handlers/typing.py b/synapse/handlers/typing.py index c2762f92c7..05879fbfc6 100644 --- a/synapse/handlers/typing.py +++ b/synapse/handlers/typing.py @@ -252,7 +252,7 @@ class TypingNotificationEventSource(object): # TODO: check if user is in room events.append(self._make_event_for(room_id)) - return (events, handler._latest_room_serial) + return defer.succeed((events, handler._latest_room_serial)) def get_current_key(self): return self.handler()._latest_room_serial diff --git a/tests/handlers/test_typing.py b/tests/handlers/test_typing.py index bf34b7ccbd..39590115e0 100644 --- a/tests/handlers/test_typing.py +++ b/tests/handlers/test_typing.py @@ -175,8 +175,9 @@ class TypingNotificationsTestCase(unittest.TestCase): ]) self.assertEquals(self.event_source.get_current_key(), 1) + events = yield self.event_source.get_new_events_for_user(self.u_apple, 0, None) self.assertEquals( - self.event_source.get_new_events_for_user(self.u_apple, 0, None)[0], + events[0], [ {"type": "m.typing", "room_id": self.room_id, @@ -237,8 +238,9 @@ class TypingNotificationsTestCase(unittest.TestCase): ]) self.assertEquals(self.event_source.get_current_key(), 1) + events = yield self.event_source.get_new_events_for_user(self.u_apple, 0, None) self.assertEquals( - self.event_source.get_new_events_for_user(self.u_apple, 0, None)[0], + events[0], [ {"type": "m.typing", "room_id": self.room_id, @@ -292,8 +294,9 @@ class TypingNotificationsTestCase(unittest.TestCase): yield put_json.await_calls() self.assertEquals(self.event_source.get_current_key(), 1) + events = yield self.event_source.get_new_events_for_user(self.u_apple, 0, None) self.assertEquals( - self.event_source.get_new_events_for_user(self.u_apple, 0, None)[0], + events[0], [ {"type": "m.typing", "room_id": self.room_id, @@ -322,8 +325,9 @@ class TypingNotificationsTestCase(unittest.TestCase): self.on_new_user_event.reset_mock() self.assertEquals(self.event_source.get_current_key(), 1) + events = yield self.event_source.get_new_events_for_user(self.u_apple, 0, None) self.assertEquals( - self.event_source.get_new_events_for_user(self.u_apple, 0, None)[0], + events[0], [ {"type": "m.typing", "room_id": self.room_id, @@ -340,8 +344,9 @@ class TypingNotificationsTestCase(unittest.TestCase): ]) self.assertEquals(self.event_source.get_current_key(), 2) + events = yield self.event_source.get_new_events_for_user(self.u_apple, 1, None) self.assertEquals( - self.event_source.get_new_events_for_user(self.u_apple, 1, None)[0], + events[0], [ {"type": "m.typing", "room_id": self.room_id, @@ -366,8 +371,9 @@ class TypingNotificationsTestCase(unittest.TestCase): self.on_new_user_event.reset_mock() self.assertEquals(self.event_source.get_current_key(), 3) + events = yield self.event_source.get_new_events_for_user(self.u_apple, 0, None) self.assertEquals( - self.event_source.get_new_events_for_user(self.u_apple, 0, None)[0], + events[0], [ {"type": "m.typing", "room_id": self.room_id, diff --git a/tests/rest/client/v1/test_typing.py b/tests/rest/client/v1/test_typing.py index 80f2ec9ddf..d04e5abda4 100644 --- a/tests/rest/client/v1/test_typing.py +++ b/tests/rest/client/v1/test_typing.py @@ -115,8 +115,9 @@ class RoomTypingTestCase(RestTestCase): self.assertEquals(200, code) self.assertEquals(self.event_source.get_current_key(), 1) + events = yield self.event_source.get_new_events_for_user(self.user_id, 0, None) self.assertEquals( - self.event_source.get_new_events_for_user(self.user_id, 0, None)[0], + events[0], [ {"type": "m.typing", "room_id": self.room_id, -- cgit 1.5.1 From 04c7f3576ec586ec16aa232794d01035390f8611 Mon Sep 17 00:00:00 2001 From: "Paul \"LeoNerd\" Evans" Date: Wed, 15 Apr 2015 23:32:11 +0100 Subject: Various minor fixes to unit-test structure around typing notifications --- tests/handlers/test_typing.py | 7 +++++++ tests/rest/client/v1/test_typing.py | 6 ++++-- 2 files changed, 11 insertions(+), 2 deletions(-) (limited to 'tests') diff --git a/tests/handlers/test_typing.py b/tests/handlers/test_typing.py index 39590115e0..91d4102fee 100644 --- a/tests/handlers/test_typing.py +++ b/tests/handlers/test_typing.py @@ -126,6 +126,13 @@ class TypingNotificationsTestCase(unittest.TestCase): return defer.succeed([]) self.room_member_handler.get_room_members = get_room_members + def get_joined_rooms_for_user(user): + if user in self.room_members: + return defer.succeed([self.room_id]) + else: + return defer.succeed([]) + self.room_member_handler.get_joined_rooms_for_user = get_joined_rooms_for_user + @defer.inlineCallbacks def fetch_room_distributions_into(room_id, localusers=None, remotedomains=None, ignore_user=None): diff --git a/tests/rest/client/v1/test_typing.py b/tests/rest/client/v1/test_typing.py index d04e5abda4..7b3bd87439 100644 --- a/tests/rest/client/v1/test_typing.py +++ b/tests/rest/client/v1/test_typing.py @@ -34,6 +34,8 @@ class RoomTypingTestCase(RestTestCase): """ Tests /rooms/$room_id/typing/$user_id REST API. """ user_id = "@sid:red" + user = UserID.from_string(user_id) + @defer.inlineCallbacks def setUp(self): self.clock = MockClock() @@ -75,7 +77,7 @@ class RoomTypingTestCase(RestTestCase): def get_room_members(room_id): if room_id == self.room_id: - return defer.succeed([UserID.from_string(self.user_id)]) + return defer.succeed([self.user]) else: return defer.succeed([]) @@ -115,7 +117,7 @@ class RoomTypingTestCase(RestTestCase): self.assertEquals(200, code) self.assertEquals(self.event_source.get_current_key(), 1) - events = yield self.event_source.get_new_events_for_user(self.user_id, 0, None) + events = yield self.event_source.get_new_events_for_user(self.user, 0, None) self.assertEquals( events[0], [ -- cgit 1.5.1 From 8a785c3006327076245428d26e5ca1634e9caeb2 Mon Sep 17 00:00:00 2001 From: "Paul \"LeoNerd\" Evans" Date: Thu, 23 Apr 2015 18:40:19 +0100 Subject: Store a list of the presence serial number at which remote users went offline, so that when we delete them from the cachemap, we can still synthesize OFFLINE events for them (SYN-261) --- synapse/handlers/presence.py | 21 +++++++++++++++++++++ tests/handlers/test_presence.py | 38 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+) (limited to 'tests') diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index 6332f50974..42fb622c48 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -135,6 +135,9 @@ class PresenceHandler(BaseHandler): self._remote_sendmap = {} # map remote users to sets of local users who're interested in them self._remote_recvmap = {} + # list of (serial, set of(userids)) tuples, ordered by serial, latest + # first + self._remote_offline_serials = [] # map any user to a UserPresenceCache self._user_cachemap = {} @@ -715,6 +718,10 @@ class PresenceHandler(BaseHandler): ) if state["presence"] == PresenceState.OFFLINE: + self._remote_offline_serials.insert( + 0, + (self._user_cachemap_latest_serial, set([user.to_string()])) + ) del self._user_cachemap[user] for poll in content.get("poll", []): @@ -856,6 +863,20 @@ class PresenceEventSource(object): # TODO(paul): limit + for serial, user_ids in presence._remote_offline_serials: + if serial < from_key: + break + + for u in user_ids: + updates.append({ + "type": "m.presence", + "content": {"user_id": u, "presence": PresenceState.OFFLINE}, + }) + # TODO(paul): For the v2 API we want to tell the client their from_key + # is too old if we fell off the end of the _remote_offline_serials + # list, and get them to invalidate+resync. In v1 we have no such + # concept so this is a best-effort result. + if updates: defer.returnValue((updates, latest_serial)) else: diff --git a/tests/handlers/test_presence.py b/tests/handlers/test_presence.py index 04eba4289e..bb497b6f09 100644 --- a/tests/handlers/test_presence.py +++ b/tests/handlers/test_presence.py @@ -878,6 +878,44 @@ class PresencePushTestCase(MockedDatastorePresenceTestCase): state ) + @defer.inlineCallbacks + def test_recv_remote_offline(self): + """ Various tests relating to SYN-261 """ + potato_set = self.handler._remote_recvmap.setdefault(self.u_potato, + set()) + potato_set.add(self.u_apple) + + self.room_members = [self.u_banana, self.u_potato] + + self.assertEquals(self.event_source.get_current_key(), 0) + + yield self.mock_federation_resource.trigger("PUT", + "/_matrix/federation/v1/send/1000000/", + _make_edu_json("elsewhere", "m.presence", + content={ + "push": [ + {"user_id": "@potato:remote", + "presence": "offline"}, + ], + } + ) + ) + + self.assertEquals(self.event_source.get_current_key(), 1) + + (events, _) = yield self.event_source.get_new_events_for_user( + self.u_apple, 0, None + ) + self.assertEquals(events, + [ + {"type": "m.presence", + "content": { + "user_id": "@potato:remote", + "presence": OFFLINE, + }} + ] + ) + @defer.inlineCallbacks def test_join_room_local(self): self.room_members = [self.u_apple, self.u_banana] -- cgit 1.5.1 From e1e5e53127540fbaa4e23fbc628113983efd767b Mon Sep 17 00:00:00 2001 From: "Paul \"LeoNerd\" Evans" Date: Thu, 23 Apr 2015 19:01:37 +0100 Subject: Remove users from the remote_offline_serials list (and clean up empty elements) when they go online again --- synapse/handlers/presence.py | 12 +++++++++++- tests/handlers/test_presence.py | 27 +++++++++++++++++++++++++++ 2 files changed, 38 insertions(+), 1 deletion(-) (limited to 'tests') diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index f929bcf853..571eacd343 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -720,14 +720,24 @@ class PresenceHandler(BaseHandler): statuscache=statuscache, ) + user_id = user.to_string() + if state["presence"] == PresenceState.OFFLINE: self._remote_offline_serials.insert( 0, - (self._user_cachemap_latest_serial, set([user.to_string()])) + (self._user_cachemap_latest_serial, set([user_id])) ) while len(self._remote_offline_serials) > MAX_OFFLINE_SERIALS: self._remote_offline_serials.pop() # remove the oldest del self._user_cachemap[user] + else: + # Remove the user from remote_offline_serials now that they're + # no longer offline + for idx, elem in enumerate(self._remote_offline_serials): + (_, user_ids) = elem + user_ids.discard(user_id) + if not user_ids: + self._remote_offline_serials.pop(idx) for poll in content.get("poll", []): user = UserID.from_string(poll) diff --git a/tests/handlers/test_presence.py b/tests/handlers/test_presence.py index bb497b6f09..9f5580c096 100644 --- a/tests/handlers/test_presence.py +++ b/tests/handlers/test_presence.py @@ -916,6 +916,33 @@ class PresencePushTestCase(MockedDatastorePresenceTestCase): ] ) + yield self.mock_federation_resource.trigger("PUT", + "/_matrix/federation/v1/send/1000001/", + _make_edu_json("elsewhere", "m.presence", + content={ + "push": [ + {"user_id": "@potato:remote", + "presence": "online"}, + ], + } + ) + ) + + self.assertEquals(self.event_source.get_current_key(), 2) + + (events, _) = yield self.event_source.get_new_events_for_user( + self.u_apple, 0, None + ) + self.assertEquals(events, + [ + {"type": "m.presence", + "content": { + "user_id": "@potato:remote", + "presence": ONLINE, + }} + ] + ) + @defer.inlineCallbacks def test_join_room_local(self): self.room_members = [self.u_apple, self.u_banana] -- cgit 1.5.1 From 43c2e8deae5f7e2b339ab5c131391231886cad09 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 5 May 2015 15:13:25 +0100 Subject: Add support for using executemany --- synapse/storage/_base.py | 54 ++++++++++++++++++++++++++++--------- synapse/storage/event_federation.py | 40 ++++++++++++++------------- synapse/storage/events.py | 46 +++++++++++++++++-------------- synapse/storage/state.py | 16 ++++++----- tests/storage/test_base.py | 4 +-- 5 files changed, 99 insertions(+), 61 deletions(-) (limited to 'tests') diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index e01c61d08d..b7c3cf03c8 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -160,18 +160,23 @@ class LoggingTransaction(object): def __setattr__(self, name, value): setattr(self.txn, name, value) - def execute(self, sql, *args, **kwargs): + def execute(self, sql, *args): + self._do_execute(self.txn.execute, sql, *args) + + def executemany(self, sql, *args): + self._do_execute(self.txn.executemany, sql, *args) + + def _do_execute(self, func, sql, *args): # TODO(paul): Maybe use 'info' and 'debug' for values? sql_logger.debug("[SQL] {%s} %s", self.name, sql) sql = self.database_engine.convert_param_style(sql) - if args and args[0]: + if args: try: sql_logger.debug( - "[SQL values] {%s} " + ", ".join(("<%r>",) * len(args[0])), - self.name, - *args[0] + "[SQL values] {%s} %r", + self.name, args[0] ) except: # Don't let logging failures stop SQL from working @@ -180,8 +185,8 @@ class LoggingTransaction(object): start = time.time() * 1000 try: - return self.txn.execute( - sql, *args, **kwargs + return func( + sql, *args ) except Exception as e: logger.debug("[SQL FAIL] {%s} %s", self.name, e) @@ -434,18 +439,41 @@ class SQLBaseStore(object): @log_function def _simple_insert_txn(self, txn, table, values): + keys, vals = zip(*values.items()) + sql = "INSERT INTO %s (%s) VALUES(%s)" % ( table, - ", ".join(k for k in values), - ", ".join("?" for k in values) + ", ".join(k for k in keys), + ", ".join("?" for _ in keys) ) - logger.debug( - "[SQL] %s Args=%s", - sql, values.values(), + txn.execute(sql, vals) + + def _simple_insert_many_txn(self, txn, table, values): + if not values: + return + + keys, vals = zip(*[ + zip( + *(sorted(i.items(), key=lambda kv: kv[0])) + ) + for i in values + if i + ]) + + for k in keys: + if k != keys[0]: + raise RuntimeError( + "All items must have the same keys" + ) + + sql = "INSERT INTO %s (%s) VALUES(%s)" % ( + table, + ", ".join(k for k in keys[0]), + ", ".join("?" for _ in keys[0]) ) - txn.execute(sql, values.values()) + txn.executemany(sql, vals) def _simple_upsert(self, table, keyvalues, values, insertion_values={}, desc="_simple_upsert", lock=True): diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 68f39bd684..0aca4ba17b 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -262,18 +262,19 @@ class EventFederationStore(SQLBaseStore): For the given event, update the event edges table and forward and backward extremities tables. """ - for e_id, _ in prev_events: - # TODO (erikj): This could be done as a bulk insert - self._simple_insert_txn( - txn, + self._simple_insert_many_txn( + txn, table="event_edges", - values={ - "event_id": event_id, - "prev_event_id": e_id, - "room_id": room_id, - "is_state": False, - }, - ) + values=[ + { + "event_id": event_id, + "prev_event_id": e_id, + "room_id": room_id, + "is_state": False, + } + for e_id, _ in prev_events + ], + ) # Update the extremities table if this is not an outlier. if not outlier: @@ -307,16 +308,17 @@ class EventFederationStore(SQLBaseStore): # Insert all the prev_events as a backwards thing, they'll get # deleted in a second if they're incorrect anyway. - for e_id, _ in prev_events: - # TODO (erikj): This could be done as a bulk insert - self._simple_insert_txn( - txn, - table="event_backward_extremities", - values={ + self._simple_insert_many_txn( + txn, + table="event_backward_extremities", + values=[ + { "event_id": e_id, "room_id": room_id, - }, - ) + } + for e_id, _ in prev_events + ], + ) # Also delete from the backwards extremities table all ones that # reference events that we have already seen diff --git a/synapse/storage/events.py b/synapse/storage/events.py index a3c260ddc4..84e446a99c 100644 --- a/synapse/storage/events.py +++ b/synapse/storage/events.py @@ -113,17 +113,19 @@ class EventsStore(SQLBaseStore): keyvalues={"room_id": event.room_id}, ) - for s in current_state: - self._simple_insert_txn( - txn, - "current_state_events", + self._simple_insert_many_txn( + txn, + "current_state_events", + [ { "event_id": s.event_id, "room_id": s.room_id, "type": s.type, "state_key": s.state_key, - }, - ) + } + for s in current_state + ], + ) if event.is_state() and is_new_state: if not backfilled and not context.rejected: @@ -296,16 +298,18 @@ class EventsStore(SQLBaseStore): txn, event.event_id, prev_event_id, alg, hash_bytes ) - for auth_id, _ in event.auth_events: - self._simple_insert_txn( - txn, - table="event_auth", - values={ + self._simple_insert_many_txn( + txn, + table="event_auth", + values=[ + { "event_id": event.event_id, "room_id": event.room_id, "auth_id": auth_id, - }, - ) + } + for auth_id, _ in event.auth_events + ], + ) (ref_alg, ref_hash_bytes) = compute_event_reference_hash(event) self._store_event_reference_hash_txn( @@ -330,17 +334,19 @@ class EventsStore(SQLBaseStore): vals, ) - for e_id, h in event.prev_state: - self._simple_insert_txn( - txn, - table="event_edges", - values={ + self._simple_insert_many_txn( + txn, + table="event_edges", + values=[ + { "event_id": event.event_id, "prev_event_id": e_id, "room_id": event.room_id, "is_state": True, - }, - ) + } + for e_id, h in event.prev_state + ], + ) if is_new_state and not context.rejected: self._simple_upsert_txn( diff --git a/synapse/storage/state.py b/synapse/storage/state.py index 7e55e8bed6..dbc0e49c1f 100644 --- a/synapse/storage/state.py +++ b/synapse/storage/state.py @@ -104,18 +104,20 @@ class StateStore(SQLBaseStore): }, ) - for state in state_events.values(): - self._simple_insert_txn( - txn, - table="state_groups_state", - values={ + self._simple_insert_many_txn( + txn, + table="state_groups_state", + values=[ + { "state_group": state_group, "room_id": state.room_id, "type": state.type, "state_key": state.state_key, "event_id": state.event_id, - }, - ) + } + for state in state_events.values() + ], + ) self._simple_insert_txn( txn, diff --git a/tests/storage/test_base.py b/tests/storage/test_base.py index a64d2b821e..8c348ecc95 100644 --- a/tests/storage/test_base.py +++ b/tests/storage/test_base.py @@ -67,7 +67,7 @@ class SQLBaseStoreTestCase(unittest.TestCase): self.mock_txn.execute.assert_called_with( "INSERT INTO tablename (columname) VALUES(?)", - ["Value"] + ("Value",) ) @defer.inlineCallbacks @@ -82,7 +82,7 @@ class SQLBaseStoreTestCase(unittest.TestCase): self.mock_txn.execute.assert_called_with( "INSERT INTO tablename (colA, colB, colC) VALUES(?, ?, ?)", - [1, 2, 3] + (1, 2, 3,) ) @defer.inlineCallbacks -- cgit 1.5.1