From 2771447c29402e1f26bb45bdb730bdd0fe02682f Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Thu, 10 Nov 2016 14:49:26 +0000 Subject: [PATCH 1/9] Store Promise instead of Response for HTTP API transactions This fixes a race whereby: - User hits an endpoint. - No cached transaction so executes main code. - User hits same endpoint. - No cache transaction so executes main code. - Main code finishes executing and caches response and returns. - Main code finishes executing and caches response and returns. This race is common in the wild when Synapse is struggling under load. This commit fixes the race by: - User hits an endpoint. - Caches the promise to execute the main code and executes main code. - User hits same endpoint. - Yields on the same promise as the first request. - Main code finishes executing and returns, unblocking both requests. --- synapse/rest/client/v1/base.py | 4 +- synapse/rest/client/v1/room.py | 82 ++++++++++---------- synapse/rest/client/v1/transactions.py | 52 ++++--------- synapse/rest/client/v2_alpha/sendtodevice.py | 18 +++-- 4 files changed, 68 insertions(+), 88 deletions(-) diff --git a/synapse/rest/client/v1/base.py b/synapse/rest/client/v1/base.py index c2a8447860..22c740c30c 100644 --- a/synapse/rest/client/v1/base.py +++ b/synapse/rest/client/v1/base.py @@ -18,7 +18,7 @@ from synapse.http.servlet import RestServlet from synapse.api.urls import CLIENT_PREFIX -from .transactions import HttpTransactionStore +from .transactions import HttpTransactionCache import re import logging @@ -59,4 +59,4 @@ class ClientV1RestServlet(RestServlet): self.hs = hs self.builder_factory = hs.get_event_builder_factory() self.auth = hs.get_v1auth() - self.txns = HttpTransactionStore() + self.txns = HttpTransactionCache() diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py index 010fbc7c32..2e919de9f3 100644 --- a/synapse/rest/client/v1/room.py +++ b/synapse/rest/client/v1/room.py @@ -56,15 +56,15 @@ class RoomCreateRestServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_PUT(self, request, txn_id): try: - defer.returnValue( - self.txns.get_client_transaction(request, txn_id) - ) + res_deferred = self.txns.get_client_transaction(request, txn_id) + res = yield res_deferred + defer.returnValue(res) except KeyError: pass - response = yield self.on_POST(request) - - self.txns.store_client_transaction(request, txn_id, response) + res_deferred = self.on_POST(request) + self.txns.store_client_transaction(request, txn_id, res_deferred) + response = yield res_deferred defer.returnValue(response) @defer.inlineCallbacks @@ -217,15 +217,15 @@ class RoomSendEventRestServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_PUT(self, request, room_id, event_type, txn_id): try: - defer.returnValue( - self.txns.get_client_transaction(request, txn_id) - ) + res_deferred = self.txns.get_client_transaction(request, txn_id) + res = yield res_deferred + defer.returnValue(res) except KeyError: pass - - response = yield self.on_POST(request, room_id, event_type, txn_id) - - self.txns.store_client_transaction(request, txn_id, response) + + res_deferred = self.on_POST(request, room_id, event_type, txn_id) + self.txns.store_client_transaction(request, txn_id, res_deferred) + response = yield res_deferred defer.returnValue(response) @@ -286,15 +286,15 @@ class JoinRoomAliasServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_PUT(self, request, room_identifier, txn_id): try: - defer.returnValue( - self.txns.get_client_transaction(request, txn_id) - ) + res_deferred = self.txns.get_client_transaction(request, txn_id) + res = yield res_deferred + defer.returnValue(res) except KeyError: pass - - response = yield self.on_POST(request, room_identifier, txn_id) - - self.txns.store_client_transaction(request, txn_id, response) + + res_deferred = self.on_POST(request, room_identifier, txn_id) + self.txns.store_client_transaction(request, txn_id, res_deferred) + response = yield res_deferred defer.returnValue(response) @@ -540,17 +540,15 @@ class RoomForgetRestServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_PUT(self, request, room_id, txn_id): try: - defer.returnValue( - self.txns.get_client_transaction(request, txn_id) - ) + res_deferred = self.txns.get_client_transaction(request, txn_id) + res = yield res_deferred + defer.returnValue(res) except KeyError: pass - - response = yield self.on_POST( - request, room_id, txn_id - ) - - self.txns.store_client_transaction(request, txn_id, response) + + res_deferred = self.on_POST(request, room_id, txn_id) + self.txns.store_client_transaction(request, txn_id, res_deferred) + response = yield res_deferred defer.returnValue(response) @@ -626,17 +624,15 @@ class RoomMembershipRestServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_PUT(self, request, room_id, membership_action, txn_id): try: - defer.returnValue( - self.txns.get_client_transaction(request, txn_id) - ) + res_deferred = self.txns.get_client_transaction(request, txn_id) + res = yield res_deferred + defer.returnValue(res) except KeyError: pass - response = yield self.on_POST( - request, room_id, membership_action, txn_id - ) - - self.txns.store_client_transaction(request, txn_id, response) + res_deferred = self.on_POST(request, room_id, membership_action, txn_id) + self.txns.store_client_transaction(request, txn_id, res_deferred) + response = yield res_deferred defer.returnValue(response) @@ -672,15 +668,15 @@ class RoomRedactEventRestServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_PUT(self, request, room_id, event_id, txn_id): try: - defer.returnValue( - self.txns.get_client_transaction(request, txn_id) - ) + res_deferred = self.txns.get_client_transaction(request, txn_id) + res = yield res_deferred + defer.returnValue(res) except KeyError: pass - response = yield self.on_POST(request, room_id, event_id, txn_id) - - self.txns.store_client_transaction(request, txn_id, response) + res_deferred = self.on_POST(request, room_id, event_id, txn_id) + self.txns.store_client_transaction(request, txn_id, res_deferred) + response = yield res_deferred defer.returnValue(response) diff --git a/synapse/rest/client/v1/transactions.py b/synapse/rest/client/v1/transactions.py index 2f2c9d0881..f5012c5f59 100644 --- a/synapse/rest/client/v1/transactions.py +++ b/synapse/rest/client/v1/transactions.py @@ -22,57 +22,35 @@ from synapse.api.auth import get_access_token_from_request logger = logging.getLogger(__name__) -# FIXME: elsewhere we use FooStore to indicate something in the storage layer... -class HttpTransactionStore(object): +class HttpTransactionCache(object): def __init__(self): - # { key : (txn_id, response) } + # { key : (txn_id, response_deferred) } self.transactions = {} - def get_response(self, key, txn_id): - """Retrieve a response for this request. - - Args: - key (str): A transaction-independent key for this request. Usually - this is a combination of the path (without the transaction id) - and the user's access token. - txn_id (str): The transaction ID for this request - Returns: - A tuple of (HTTP response code, response content) or None. - """ + def _get_response(self, key, txn_id): try: - logger.debug("get_response TxnId: %s", txn_id) - (last_txn_id, response) = self.transactions[key] + (last_txn_id, response_deferred) = self.transactions[key] if txn_id == last_txn_id: logger.info("get_response: Returning a response for %s", txn_id) - return response + return response_deferred except KeyError: pass return None - def store_response(self, key, txn_id, response): - """Stores an HTTP response tuple. + def _store_response(self, key, txn_id, response_deferred): + self.transactions[key] = (txn_id, response_deferred) - Args: - key (str): A transaction-independent key for this request. Usually - this is a combination of the path (without the transaction id) - and the user's access token. - txn_id (str): The transaction ID for this request. - response (tuple): A tuple of (HTTP response code, response content) - """ - logger.debug("store_response TxnId: %s", txn_id) - self.transactions[key] = (txn_id, response) - - def store_client_transaction(self, request, txn_id, response): - """Stores the request/response pair of an HTTP transaction. + def store_client_transaction(self, request, txn_id, response_deferred): + """Stores the request/Promise pair of an HTTP transaction. Args: request (twisted.web.http.Request): The twisted HTTP request. This request must have the transaction ID as the last path segment. - response (tuple): A tuple of (response code, response dict) + response_deferred (Promise): A tuple of (response code, response dict) txn_id (str): The transaction ID for this request. """ - self.store_response(self._get_key(request), txn_id, response) + self._store_response(self._get_key(request), txn_id, response_deferred) def get_client_transaction(self, request, txn_id): """Retrieves a stored response if there was one. @@ -82,14 +60,14 @@ class HttpTransactionStore(object): request must have the transaction ID as the last path segment. txn_id (str): The transaction ID for this request. Returns: - The response tuple. + Promise: Resolves to the response tuple. Raises: KeyError if the transaction was not found. """ - response = self.get_response(self._get_key(request), txn_id) - if response is None: + response_deferred = self._get_response(self._get_key(request), txn_id) + if response_deferred is None: raise KeyError("Transaction not found.") - return response + return response_deferred def _get_key(self, request): token = get_access_token_from_request(request) diff --git a/synapse/rest/client/v2_alpha/sendtodevice.py b/synapse/rest/client/v2_alpha/sendtodevice.py index 5975164b37..7c800ca895 100644 --- a/synapse/rest/client/v2_alpha/sendtodevice.py +++ b/synapse/rest/client/v2_alpha/sendtodevice.py @@ -19,7 +19,7 @@ from twisted.internet import defer from synapse.http import servlet from synapse.http.servlet import parse_json_object_from_request -from synapse.rest.client.v1.transactions import HttpTransactionStore +from synapse.rest.client.v1.transactions import HttpTransactionCache from ._base import client_v2_patterns @@ -40,18 +40,25 @@ class SendToDeviceRestServlet(servlet.RestServlet): super(SendToDeviceRestServlet, self).__init__() self.hs = hs self.auth = hs.get_auth() - self.txns = HttpTransactionStore() + self.txns = HttpTransactionCache() self.device_message_handler = hs.get_device_message_handler() @defer.inlineCallbacks def on_PUT(self, request, message_type, txn_id): try: - defer.returnValue( - self.txns.get_client_transaction(request, txn_id) - ) + res_deferred = self.txns.get_client_transaction(request, txn_id) + res = yield res_deferred + defer.returnValue(res) except KeyError: pass + + res_deferred = self._put(request, message_type, txn_id) + self.txns.store_client_transaction(request, txn_id, res_deferred) + res = yield res_deferred + defer.returnValue(res) + @defer.inlineCallbacks + def _put(self, request, message_type, txn_id): requester = yield self.auth.get_user_by_req(request) content = parse_json_object_from_request(request) @@ -63,7 +70,6 @@ class SendToDeviceRestServlet(servlet.RestServlet): ) response = (200, {}) - self.txns.store_client_transaction(request, txn_id, response) defer.returnValue(response) From 8a8ad46f48525812c983dadcd9e2757cebaaa6cf Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Thu, 10 Nov 2016 15:22:11 +0000 Subject: [PATCH 2/9] Flake8 --- synapse/rest/client/v1/room.py | 6 +++--- synapse/rest/client/v2_alpha/sendtodevice.py | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py index 2e919de9f3..caa779ffaf 100644 --- a/synapse/rest/client/v1/room.py +++ b/synapse/rest/client/v1/room.py @@ -222,7 +222,7 @@ class RoomSendEventRestServlet(ClientV1RestServlet): defer.returnValue(res) except KeyError: pass - + res_deferred = self.on_POST(request, room_id, event_type, txn_id) self.txns.store_client_transaction(request, txn_id, res_deferred) response = yield res_deferred @@ -291,7 +291,7 @@ class JoinRoomAliasServlet(ClientV1RestServlet): defer.returnValue(res) except KeyError: pass - + res_deferred = self.on_POST(request, room_identifier, txn_id) self.txns.store_client_transaction(request, txn_id, res_deferred) response = yield res_deferred @@ -545,7 +545,7 @@ class RoomForgetRestServlet(ClientV1RestServlet): defer.returnValue(res) except KeyError: pass - + res_deferred = self.on_POST(request, room_id, txn_id) self.txns.store_client_transaction(request, txn_id, res_deferred) response = yield res_deferred diff --git a/synapse/rest/client/v2_alpha/sendtodevice.py b/synapse/rest/client/v2_alpha/sendtodevice.py index 7c800ca895..4d13e793ac 100644 --- a/synapse/rest/client/v2_alpha/sendtodevice.py +++ b/synapse/rest/client/v2_alpha/sendtodevice.py @@ -51,7 +51,7 @@ class SendToDeviceRestServlet(servlet.RestServlet): defer.returnValue(res) except KeyError: pass - + res_deferred = self._put(request, message_type, txn_id) self.txns.store_client_transaction(request, txn_id, res_deferred) res = yield res_deferred From c7daf3136c8ae024187cd50530d3ee90b1385b13 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Fri, 11 Nov 2016 14:13:32 +0000 Subject: [PATCH 3/9] Use observable deferreds because they are sane --- synapse/rest/client/v2_alpha/sendtodevice.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/synapse/rest/client/v2_alpha/sendtodevice.py b/synapse/rest/client/v2_alpha/sendtodevice.py index 4d13e793ac..31167ba535 100644 --- a/synapse/rest/client/v2_alpha/sendtodevice.py +++ b/synapse/rest/client/v2_alpha/sendtodevice.py @@ -20,6 +20,7 @@ from twisted.internet import defer from synapse.http import servlet from synapse.http.servlet import parse_json_object_from_request from synapse.rest.client.v1.transactions import HttpTransactionCache +from synapse.util.async import ObservableDeferred from ._base import client_v2_patterns @@ -47,14 +48,14 @@ class SendToDeviceRestServlet(servlet.RestServlet): def on_PUT(self, request, message_type, txn_id): try: res_deferred = self.txns.get_client_transaction(request, txn_id) - res = yield res_deferred + res = yield res_deferred.observe() defer.returnValue(res) except KeyError: pass - res_deferred = self._put(request, message_type, txn_id) + res_deferred = ObservableDeferred(self._put(request, message_type, txn_id)) self.txns.store_client_transaction(request, txn_id, res_deferred) - res = yield res_deferred + res = yield res_deferred.observe() defer.returnValue(res) @defer.inlineCallbacks From 42c43cfafd3e0471e4e0f6fb05e951290783ba1f Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Fri, 11 Nov 2016 14:54:10 +0000 Subject: [PATCH 4/9] Use ObservableDeferreds instead of Deferreds as they behave as intended --- synapse/rest/client/v1/room.py | 39 +++++++++++++------------- synapse/rest/client/v1/transactions.py | 22 +++++++-------- 2 files changed, 31 insertions(+), 30 deletions(-) diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py index caa779ffaf..47c4b84b73 100644 --- a/synapse/rest/client/v1/room.py +++ b/synapse/rest/client/v1/room.py @@ -22,6 +22,7 @@ from synapse.streams.config import PaginationConfig from synapse.api.constants import EventTypes, Membership from synapse.api.filtering import Filter from synapse.types import UserID, RoomID, RoomAlias +from synapse.util.async import ObservableDeferred from synapse.events.utils import serialize_event, format_event_for_client_v2 from synapse.http.servlet import ( parse_json_object_from_request, parse_string, parse_integer @@ -57,14 +58,14 @@ class RoomCreateRestServlet(ClientV1RestServlet): def on_PUT(self, request, txn_id): try: res_deferred = self.txns.get_client_transaction(request, txn_id) - res = yield res_deferred + res = yield res_deferred.observe() defer.returnValue(res) except KeyError: pass - res_deferred = self.on_POST(request) + res_deferred = ObservableDeferred(self.on_POST(request)) self.txns.store_client_transaction(request, txn_id, res_deferred) - response = yield res_deferred + response = yield res_deferred.observe() defer.returnValue(response) @defer.inlineCallbacks @@ -218,14 +219,14 @@ class RoomSendEventRestServlet(ClientV1RestServlet): def on_PUT(self, request, room_id, event_type, txn_id): try: res_deferred = self.txns.get_client_transaction(request, txn_id) - res = yield res_deferred + res = yield res_deferred.observe() defer.returnValue(res) except KeyError: pass - res_deferred = self.on_POST(request, room_id, event_type, txn_id) + res_deferred = ObservableDeferred(self.on_POST(request, room_id, event_type, txn_id)) self.txns.store_client_transaction(request, txn_id, res_deferred) - response = yield res_deferred + response = yield res_deferred.observe() defer.returnValue(response) @@ -287,14 +288,14 @@ class JoinRoomAliasServlet(ClientV1RestServlet): def on_PUT(self, request, room_identifier, txn_id): try: res_deferred = self.txns.get_client_transaction(request, txn_id) - res = yield res_deferred + res = yield res_deferred.observe() defer.returnValue(res) except KeyError: pass - res_deferred = self.on_POST(request, room_identifier, txn_id) + res_deferred = ObservableDeferred(self.on_POST(request, room_identifier, txn_id)) self.txns.store_client_transaction(request, txn_id, res_deferred) - response = yield res_deferred + response = yield res_deferred.observe() defer.returnValue(response) @@ -541,14 +542,14 @@ class RoomForgetRestServlet(ClientV1RestServlet): def on_PUT(self, request, room_id, txn_id): try: res_deferred = self.txns.get_client_transaction(request, txn_id) - res = yield res_deferred + res = yield res_deferred.observe() defer.returnValue(res) except KeyError: pass - res_deferred = self.on_POST(request, room_id, txn_id) + res_deferred = ObservableDeferred(self.on_POST(request, room_id, txn_id)) self.txns.store_client_transaction(request, txn_id, res_deferred) - response = yield res_deferred + response = yield res_deferred.observe() defer.returnValue(response) @@ -624,15 +625,15 @@ class RoomMembershipRestServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_PUT(self, request, room_id, membership_action, txn_id): try: - res_deferred = self.txns.get_client_transaction(request, txn_id) - res = yield res_deferred + res_deferred = ObservableDeferred(self.txns.get_client_transaction(request, txn_id)) + res = yield res_deferred.observe() defer.returnValue(res) except KeyError: pass - res_deferred = self.on_POST(request, room_id, membership_action, txn_id) + res_deferred = ObservableDeffself.on_POST(request, room_id, membership_action, txn_id) self.txns.store_client_transaction(request, txn_id, res_deferred) - response = yield res_deferred + response = yield res_deferred.observe() defer.returnValue(response) @@ -669,14 +670,14 @@ class RoomRedactEventRestServlet(ClientV1RestServlet): def on_PUT(self, request, room_id, event_id, txn_id): try: res_deferred = self.txns.get_client_transaction(request, txn_id) - res = yield res_deferred + res = yield res_deferred.observe() defer.returnValue(res) except KeyError: pass - res_deferred = self.on_POST(request, room_id, event_id, txn_id) + res_deferred = ObservableDeferred(self.on_POST(request, room_id, event_id, txn_id)) self.txns.store_client_transaction(request, txn_id, res_deferred) - response = yield res_deferred + response = yield res_deferred.observe() defer.returnValue(response) diff --git a/synapse/rest/client/v1/transactions.py b/synapse/rest/client/v1/transactions.py index f5012c5f59..774430458a 100644 --- a/synapse/rest/client/v1/transactions.py +++ b/synapse/rest/client/v1/transactions.py @@ -25,32 +25,32 @@ logger = logging.getLogger(__name__) class HttpTransactionCache(object): def __init__(self): - # { key : (txn_id, response_deferred) } + # { key : (txn_id, res_observ_defer) } self.transactions = {} def _get_response(self, key, txn_id): try: - (last_txn_id, response_deferred) = self.transactions[key] + (last_txn_id, res_observ_defer) = self.transactions[key] if txn_id == last_txn_id: logger.info("get_response: Returning a response for %s", txn_id) - return response_deferred + return res_observ_defer except KeyError: pass return None - def _store_response(self, key, txn_id, response_deferred): - self.transactions[key] = (txn_id, response_deferred) + def _store_response(self, key, txn_id, res_observ_defer): + self.transactions[key] = (txn_id, res_observ_defer) - def store_client_transaction(self, request, txn_id, response_deferred): + def store_client_transaction(self, request, txn_id, res_observ_defer): """Stores the request/Promise pair of an HTTP transaction. Args: request (twisted.web.http.Request): The twisted HTTP request. This request must have the transaction ID as the last path segment. - response_deferred (Promise): A tuple of (response code, response dict) + res_observ_defer (Promise): A tuple of (response code, response dict) txn_id (str): The transaction ID for this request. """ - self._store_response(self._get_key(request), txn_id, response_deferred) + self._store_response(self._get_key(request), txn_id, res_observ_defer) def get_client_transaction(self, request, txn_id): """Retrieves a stored response if there was one. @@ -64,10 +64,10 @@ class HttpTransactionCache(object): Raises: KeyError if the transaction was not found. """ - response_deferred = self._get_response(self._get_key(request), txn_id) - if response_deferred is None: + res_observ_defer = self._get_response(self._get_key(request), txn_id) + if res_observ_defer is None: raise KeyError("Transaction not found.") - return response_deferred + return res_observ_defer def _get_key(self, request): token = get_access_token_from_request(request) From a88bc67f88bf5b932283372818c7e33bedfbaa0b Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Fri, 11 Nov 2016 15:02:29 +0000 Subject: [PATCH 5/9] Flake8 and fix whoopsie --- synapse/rest/client/v1/room.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py index 47c4b84b73..a866615f71 100644 --- a/synapse/rest/client/v1/room.py +++ b/synapse/rest/client/v1/room.py @@ -224,7 +224,9 @@ class RoomSendEventRestServlet(ClientV1RestServlet): except KeyError: pass - res_deferred = ObservableDeferred(self.on_POST(request, room_id, event_type, txn_id)) + res_deferred = ObservableDeferred( + self.on_POST(request, room_id, event_type, txn_id) + ) self.txns.store_client_transaction(request, txn_id, res_deferred) response = yield res_deferred.observe() defer.returnValue(response) @@ -625,13 +627,13 @@ class RoomMembershipRestServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_PUT(self, request, room_id, membership_action, txn_id): try: - res_deferred = ObservableDeferred(self.txns.get_client_transaction(request, txn_id)) + res_deferred = self.txns.get_client_transaction(request, txn_id) res = yield res_deferred.observe() defer.returnValue(res) except KeyError: pass - res_deferred = ObservableDeffself.on_POST(request, room_id, membership_action, txn_id) + res_deferred = ObservableDeferred(self.on_POST(request, room_id, membership_action, txn_id)) self.txns.store_client_transaction(request, txn_id, res_deferred) response = yield res_deferred.observe() defer.returnValue(response) @@ -675,7 +677,9 @@ class RoomRedactEventRestServlet(ClientV1RestServlet): except KeyError: pass - res_deferred = ObservableDeferred(self.on_POST(request, room_id, event_id, txn_id)) + res_deferred = ObservableDeferred( + self.on_POST(request, room_id, event_id, txn_id) + ) self.txns.store_client_transaction(request, txn_id, res_deferred) response = yield res_deferred.observe() defer.returnValue(response) From f6c48802f5d01cbbc5fbd9ac151b84a80a2e3ac3 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Fri, 11 Nov 2016 15:08:24 +0000 Subject: [PATCH 6/9] More flake8 --- synapse/rest/client/v1/room.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py index a866615f71..7af4077721 100644 --- a/synapse/rest/client/v1/room.py +++ b/synapse/rest/client/v1/room.py @@ -633,7 +633,9 @@ class RoomMembershipRestServlet(ClientV1RestServlet): except KeyError: pass - res_deferred = ObservableDeferred(self.on_POST(request, room_id, membership_action, txn_id)) + res_deferred = ObservableDeferred( + self.on_POST(request, room_id, membership_action, txn_id) + ) self.txns.store_client_transaction(request, txn_id, res_deferred) response = yield res_deferred.observe() defer.returnValue(response) From 8ecaff51a147948f977e745bace697ffcba6595b Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Fri, 11 Nov 2016 17:47:03 +0000 Subject: [PATCH 7/9] Review comments --- synapse/rest/client/transactions.py | 85 +++++++++++++++++ synapse/rest/client/v1/base.py | 3 +- synapse/rest/client/v1/room.py | 97 ++++++-------------- synapse/rest/client/v1/transactions.py | 75 --------------- synapse/rest/client/v2_alpha/sendtodevice.py | 17 +--- 5 files changed, 119 insertions(+), 158 deletions(-) create mode 100644 synapse/rest/client/transactions.py delete mode 100644 synapse/rest/client/v1/transactions.py diff --git a/synapse/rest/client/transactions.py b/synapse/rest/client/transactions.py new file mode 100644 index 0000000000..1db972a378 --- /dev/null +++ b/synapse/rest/client/transactions.py @@ -0,0 +1,85 @@ +# -*- coding: utf-8 -*- +# Copyright 2014-2016 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. + +"""This module contains logic for storing HTTP PUT transactions. This is used +to ensure idempotency when performing PUTs using the REST API.""" +import logging + +from synapse.api.auth import get_access_token_from_request +from synapse.util.async import ObservableDeferred + +logger = logging.getLogger(__name__) + + +def get_transaction_key(request): + """A helper function which returns a transaction key that can be used + with TransactionCache for idempotent requests. + + Idempotency is based on the returned key being the same for separate + requests to the same endpoint. The key is formed from the HTTP request + path and the access_token for the requesting user. + + Args: + request (twisted.web.http.Request): The incoming request. Must + contain an access_token. + Returns: + str: A transaction key + """ + token = get_access_token_from_request(request) + return request.path + "/" + token + + +class HttpTransactionCache(object): + + def __init__(self): + self.transactions = { + # $txn_key: ObservableDeferred<(res_code, res_json_body)> + } + + def fetch_or_execute_request(self, request, fn, *args, **kwargs): + """A helper function for fetch_or_execute which extracts + a transaction key from the given request. + + See: + fetch_or_execute + """ + return self.fetch_or_execute( + get_transaction_key(request), fn, *args, **kwargs + ) + + def fetch_or_execute(self, txn_key, fn, *args, **kwargs): + """Fetches the response for this transaction, or executes the given function + to produce a response for this transaction. + + Args: + txn_key (str): A key to ensure idempotency should fetch_or_execute be + called again at a later point in time. + fn (function): A function which returns a tuple of + (response_code, response_dict)d + *args: Arguments to pass to fn. + **kwargs: Keyword arguments to pass to fn. + Returns: + synapse.util.async.ObservableDeferred which resolves to a tuple + of (response_code, response_dict). + """ + try: + return self.transactions[txn_key] + except KeyError: + pass # execute the function instead. + + deferred = fn(*args, **kwargs) + observable = ObservableDeferred(deferred) + self.transactions[txn_key] = observable + return observable diff --git a/synapse/rest/client/v1/base.py b/synapse/rest/client/v1/base.py index 22c740c30c..07ff5b218c 100644 --- a/synapse/rest/client/v1/base.py +++ b/synapse/rest/client/v1/base.py @@ -18,7 +18,8 @@ from synapse.http.servlet import RestServlet from synapse.api.urls import CLIENT_PREFIX -from .transactions import HttpTransactionCache +from synapse.rest.client.transactions import HttpTransactionCache + import re import logging diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py index 7af4077721..0622e64380 100644 --- a/synapse/rest/client/v1/room.py +++ b/synapse/rest/client/v1/room.py @@ -22,7 +22,6 @@ from synapse.streams.config import PaginationConfig from synapse.api.constants import EventTypes, Membership from synapse.api.filtering import Filter from synapse.types import UserID, RoomID, RoomAlias -from synapse.util.async import ObservableDeferred from synapse.events.utils import serialize_event, format_event_for_client_v2 from synapse.http.servlet import ( parse_json_object_from_request, parse_string, parse_integer @@ -56,17 +55,11 @@ class RoomCreateRestServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_PUT(self, request, txn_id): - try: - res_deferred = self.txns.get_client_transaction(request, txn_id) - res = yield res_deferred.observe() - defer.returnValue(res) - except KeyError: - pass - - res_deferred = ObservableDeferred(self.on_POST(request)) - self.txns.store_client_transaction(request, txn_id, res_deferred) - response = yield res_deferred.observe() - defer.returnValue(response) + observable = self.txns.fetch_or_execute_request( + request, self.on_POST, request + ) + res = yield observable.observe() + defer.returnValue(res) @defer.inlineCallbacks def on_POST(self, request): @@ -217,19 +210,11 @@ class RoomSendEventRestServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_PUT(self, request, room_id, event_type, txn_id): - try: - res_deferred = self.txns.get_client_transaction(request, txn_id) - res = yield res_deferred.observe() - defer.returnValue(res) - except KeyError: - pass - - res_deferred = ObservableDeferred( - self.on_POST(request, room_id, event_type, txn_id) + observable = self.txns.fetch_or_execute_request( + request, self.on_POST, request, room_id, event_type, txn_id ) - self.txns.store_client_transaction(request, txn_id, res_deferred) - response = yield res_deferred.observe() - defer.returnValue(response) + res = yield observable.observe() + defer.returnValue(res) # TODO: Needs unit testing for room ID + alias joins @@ -288,17 +273,11 @@ class JoinRoomAliasServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_PUT(self, request, room_identifier, txn_id): - try: - res_deferred = self.txns.get_client_transaction(request, txn_id) - res = yield res_deferred.observe() - defer.returnValue(res) - except KeyError: - pass - - res_deferred = ObservableDeferred(self.on_POST(request, room_identifier, txn_id)) - self.txns.store_client_transaction(request, txn_id, res_deferred) - response = yield res_deferred.observe() - defer.returnValue(response) + observable = self.txns.fetch_or_execute_request( + request, self.on_POST, request, room_identifier, txn_id + ) + res = yield observable.observe() + defer.returnValue(res) # TODO: Needs unit testing @@ -542,17 +521,11 @@ class RoomForgetRestServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_PUT(self, request, room_id, txn_id): - try: - res_deferred = self.txns.get_client_transaction(request, txn_id) - res = yield res_deferred.observe() - defer.returnValue(res) - except KeyError: - pass - - res_deferred = ObservableDeferred(self.on_POST(request, room_id, txn_id)) - self.txns.store_client_transaction(request, txn_id, res_deferred) - response = yield res_deferred.observe() - defer.returnValue(response) + observable = self.txns.fetch_or_execute_request( + request, self.on_POST, request, room_id, txn_id + ) + res = yield observable.observe() + defer.returnValue(res) # TODO: Needs unit testing @@ -626,19 +599,11 @@ class RoomMembershipRestServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_PUT(self, request, room_id, membership_action, txn_id): - try: - res_deferred = self.txns.get_client_transaction(request, txn_id) - res = yield res_deferred.observe() - defer.returnValue(res) - except KeyError: - pass - - res_deferred = ObservableDeferred( - self.on_POST(request, room_id, membership_action, txn_id) + observable = self.txns.fetch_or_execute_request( + request, self.on_POST, request, room_id, membership_action, txn_id ) - self.txns.store_client_transaction(request, txn_id, res_deferred) - response = yield res_deferred.observe() - defer.returnValue(response) + res = yield observable.observe() + defer.returnValue(res) class RoomRedactEventRestServlet(ClientV1RestServlet): @@ -672,19 +637,11 @@ class RoomRedactEventRestServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_PUT(self, request, room_id, event_id, txn_id): - try: - res_deferred = self.txns.get_client_transaction(request, txn_id) - res = yield res_deferred.observe() - defer.returnValue(res) - except KeyError: - pass - - res_deferred = ObservableDeferred( - self.on_POST(request, room_id, event_id, txn_id) + observable = self.txns.fetch_or_execute_request( + request, self.on_POST, request, room_id, event_id, txn_id ) - self.txns.store_client_transaction(request, txn_id, res_deferred) - response = yield res_deferred.observe() - defer.returnValue(response) + res = yield observable.observe() + defer.returnValue(res) class RoomTypingRestServlet(ClientV1RestServlet): diff --git a/synapse/rest/client/v1/transactions.py b/synapse/rest/client/v1/transactions.py deleted file mode 100644 index 774430458a..0000000000 --- a/synapse/rest/client/v1/transactions.py +++ /dev/null @@ -1,75 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2014-2016 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. - -"""This module contains logic for storing HTTP PUT transactions. This is used -to ensure idempotency when performing PUTs using the REST API.""" -import logging - -from synapse.api.auth import get_access_token_from_request - -logger = logging.getLogger(__name__) - - -class HttpTransactionCache(object): - - def __init__(self): - # { key : (txn_id, res_observ_defer) } - self.transactions = {} - - def _get_response(self, key, txn_id): - try: - (last_txn_id, res_observ_defer) = self.transactions[key] - if txn_id == last_txn_id: - logger.info("get_response: Returning a response for %s", txn_id) - return res_observ_defer - except KeyError: - pass - return None - - def _store_response(self, key, txn_id, res_observ_defer): - self.transactions[key] = (txn_id, res_observ_defer) - - def store_client_transaction(self, request, txn_id, res_observ_defer): - """Stores the request/Promise pair of an HTTP transaction. - - Args: - request (twisted.web.http.Request): The twisted HTTP request. This - request must have the transaction ID as the last path segment. - res_observ_defer (Promise): A tuple of (response code, response dict) - txn_id (str): The transaction ID for this request. - """ - self._store_response(self._get_key(request), txn_id, res_observ_defer) - - def get_client_transaction(self, request, txn_id): - """Retrieves a stored response if there was one. - - Args: - request (twisted.web.http.Request): The twisted HTTP request. This - request must have the transaction ID as the last path segment. - txn_id (str): The transaction ID for this request. - Returns: - Promise: Resolves to the response tuple. - Raises: - KeyError if the transaction was not found. - """ - res_observ_defer = self._get_response(self._get_key(request), txn_id) - if res_observ_defer is None: - raise KeyError("Transaction not found.") - return res_observ_defer - - def _get_key(self, request): - token = get_access_token_from_request(request) - path_without_txn_id = request.path.rsplit("/", 1)[0] - return path_without_txn_id + "/" + token diff --git a/synapse/rest/client/v2_alpha/sendtodevice.py b/synapse/rest/client/v2_alpha/sendtodevice.py index 31167ba535..2ce038c6cd 100644 --- a/synapse/rest/client/v2_alpha/sendtodevice.py +++ b/synapse/rest/client/v2_alpha/sendtodevice.py @@ -19,8 +19,7 @@ from twisted.internet import defer from synapse.http import servlet from synapse.http.servlet import parse_json_object_from_request -from synapse.rest.client.v1.transactions import HttpTransactionCache -from synapse.util.async import ObservableDeferred +from synapse.rest.client.transactions import HttpTransactionCache from ._base import client_v2_patterns @@ -46,16 +45,10 @@ class SendToDeviceRestServlet(servlet.RestServlet): @defer.inlineCallbacks def on_PUT(self, request, message_type, txn_id): - try: - res_deferred = self.txns.get_client_transaction(request, txn_id) - res = yield res_deferred.observe() - defer.returnValue(res) - except KeyError: - pass - - res_deferred = ObservableDeferred(self._put(request, message_type, txn_id)) - self.txns.store_client_transaction(request, txn_id, res_deferred) - res = yield res_deferred.observe() + observable = self.txns.fetch_or_execute_request( + request, self._put, request, message_type, txn_id + ) + res = yield observable.observe() defer.returnValue(res) @defer.inlineCallbacks From af4a1bac5088e8083f55eea05f6fad44208a3a51 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Mon, 14 Nov 2016 09:52:41 +0000 Subject: [PATCH 8/9] Move .observe() up to the cache to make things neater --- synapse/rest/client/transactions.py | 9 +++--- synapse/rest/client/v1/room.py | 30 ++++---------------- synapse/rest/client/v2_alpha/sendtodevice.py | 5 +--- 3 files changed, 11 insertions(+), 33 deletions(-) diff --git a/synapse/rest/client/transactions.py b/synapse/rest/client/transactions.py index 1db972a378..8d69e12d36 100644 --- a/synapse/rest/client/transactions.py +++ b/synapse/rest/client/transactions.py @@ -67,19 +67,18 @@ class HttpTransactionCache(object): txn_key (str): A key to ensure idempotency should fetch_or_execute be called again at a later point in time. fn (function): A function which returns a tuple of - (response_code, response_dict)d + (response_code, response_dict). *args: Arguments to pass to fn. **kwargs: Keyword arguments to pass to fn. Returns: - synapse.util.async.ObservableDeferred which resolves to a tuple - of (response_code, response_dict). + Deferred which resolves to a tuple of (response_code, response_dict). """ try: - return self.transactions[txn_key] + return self.transactions[txn_key].observe() except KeyError: pass # execute the function instead. deferred = fn(*args, **kwargs) observable = ObservableDeferred(deferred) self.transactions[txn_key] = observable - return observable + return observable.observe() diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py index 0622e64380..3fb1f2deb3 100644 --- a/synapse/rest/client/v1/room.py +++ b/synapse/rest/client/v1/room.py @@ -53,13 +53,10 @@ class RoomCreateRestServlet(ClientV1RestServlet): client_path_patterns("/createRoom(?:/.*)?$"), self.on_OPTIONS) - @defer.inlineCallbacks def on_PUT(self, request, txn_id): - observable = self.txns.fetch_or_execute_request( + return self.txns.fetch_or_execute_request( request, self.on_POST, request ) - res = yield observable.observe() - defer.returnValue(res) @defer.inlineCallbacks def on_POST(self, request): @@ -208,13 +205,10 @@ class RoomSendEventRestServlet(ClientV1RestServlet): def on_GET(self, request, room_id, event_type, txn_id): return (200, "Not implemented") - @defer.inlineCallbacks def on_PUT(self, request, room_id, event_type, txn_id): - observable = self.txns.fetch_or_execute_request( + return self.txns.fetch_or_execute_request( request, self.on_POST, request, room_id, event_type, txn_id ) - res = yield observable.observe() - defer.returnValue(res) # TODO: Needs unit testing for room ID + alias joins @@ -271,13 +265,10 @@ class JoinRoomAliasServlet(ClientV1RestServlet): defer.returnValue((200, {"room_id": room_id})) - @defer.inlineCallbacks def on_PUT(self, request, room_identifier, txn_id): - observable = self.txns.fetch_or_execute_request( + return self.txns.fetch_or_execute_request( request, self.on_POST, request, room_identifier, txn_id ) - res = yield observable.observe() - defer.returnValue(res) # TODO: Needs unit testing @@ -519,13 +510,10 @@ class RoomForgetRestServlet(ClientV1RestServlet): defer.returnValue((200, {})) - @defer.inlineCallbacks def on_PUT(self, request, room_id, txn_id): - observable = self.txns.fetch_or_execute_request( + return self.txns.fetch_or_execute_request( request, self.on_POST, request, room_id, txn_id ) - res = yield observable.observe() - defer.returnValue(res) # TODO: Needs unit testing @@ -597,13 +585,10 @@ class RoomMembershipRestServlet(ClientV1RestServlet): return False return True - @defer.inlineCallbacks def on_PUT(self, request, room_id, membership_action, txn_id): - observable = self.txns.fetch_or_execute_request( + return self.txns.fetch_or_execute_request( request, self.on_POST, request, room_id, membership_action, txn_id ) - res = yield observable.observe() - defer.returnValue(res) class RoomRedactEventRestServlet(ClientV1RestServlet): @@ -635,13 +620,10 @@ class RoomRedactEventRestServlet(ClientV1RestServlet): defer.returnValue((200, {"event_id": event.event_id})) - @defer.inlineCallbacks def on_PUT(self, request, room_id, event_id, txn_id): - observable = self.txns.fetch_or_execute_request( + return self.txns.fetch_or_execute_request( request, self.on_POST, request, room_id, event_id, txn_id ) - res = yield observable.observe() - defer.returnValue(res) class RoomTypingRestServlet(ClientV1RestServlet): diff --git a/synapse/rest/client/v2_alpha/sendtodevice.py b/synapse/rest/client/v2_alpha/sendtodevice.py index 2ce038c6cd..2187350d42 100644 --- a/synapse/rest/client/v2_alpha/sendtodevice.py +++ b/synapse/rest/client/v2_alpha/sendtodevice.py @@ -43,13 +43,10 @@ class SendToDeviceRestServlet(servlet.RestServlet): self.txns = HttpTransactionCache() self.device_message_handler = hs.get_device_message_handler() - @defer.inlineCallbacks def on_PUT(self, request, message_type, txn_id): - observable = self.txns.fetch_or_execute_request( + return self.txns.fetch_or_execute_request( request, self._put, request, message_type, txn_id ) - res = yield observable.observe() - defer.returnValue(res) @defer.inlineCallbacks def _put(self, request, message_type, txn_id): From 3991b4cbdb5f5fbdf61ad6efa879b3881143c214 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Mon, 14 Nov 2016 11:19:24 +0000 Subject: [PATCH 9/9] Clean transactions based on time. Add HttpTransactionCache tests. --- synapse/rest/client/transactions.py | 24 +++++-- synapse/rest/client/v1/base.py | 2 +- synapse/rest/client/v2_alpha/sendtodevice.py | 2 +- synapse/util/__init__.py | 10 ++- tests/rest/client/test_transactions.py | 69 ++++++++++++++++++++ 5 files changed, 99 insertions(+), 8 deletions(-) create mode 100644 tests/rest/client/test_transactions.py diff --git a/synapse/rest/client/transactions.py b/synapse/rest/client/transactions.py index 8d69e12d36..351170edbc 100644 --- a/synapse/rest/client/transactions.py +++ b/synapse/rest/client/transactions.py @@ -41,12 +41,19 @@ def get_transaction_key(request): return request.path + "/" + token +CLEANUP_PERIOD_MS = 1000 * 60 * 30 # 30 mins + + class HttpTransactionCache(object): - def __init__(self): + def __init__(self, clock): + self.clock = clock self.transactions = { - # $txn_key: ObservableDeferred<(res_code, res_json_body)> + # $txn_key: (ObservableDeferred<(res_code, res_json_body)>, timestamp) } + # Try to clean entries every 30 mins. This means entries will exist + # for at *LEAST* 30 mins, and at *MOST* 60 mins. + self.cleaner = self.clock.looping_call(self._cleanup, CLEANUP_PERIOD_MS) def fetch_or_execute_request(self, request, fn, *args, **kwargs): """A helper function for fetch_or_execute which extracts @@ -74,11 +81,18 @@ class HttpTransactionCache(object): Deferred which resolves to a tuple of (response_code, response_dict). """ try: - return self.transactions[txn_key].observe() - except KeyError: + return self.transactions[txn_key][0].observe() + except (KeyError, IndexError): pass # execute the function instead. deferred = fn(*args, **kwargs) observable = ObservableDeferred(deferred) - self.transactions[txn_key] = observable + self.transactions[txn_key] = (observable, self.clock.time_msec()) return observable.observe() + + def _cleanup(self): + now = self.clock.time_msec() + for key in self.transactions.keys(): + ts = self.transactions[key][1] + if now > (ts + CLEANUP_PERIOD_MS): # after cleanup period + del self.transactions[key] diff --git a/synapse/rest/client/v1/base.py b/synapse/rest/client/v1/base.py index 07ff5b218c..c7aa0bbf59 100644 --- a/synapse/rest/client/v1/base.py +++ b/synapse/rest/client/v1/base.py @@ -60,4 +60,4 @@ class ClientV1RestServlet(RestServlet): self.hs = hs self.builder_factory = hs.get_event_builder_factory() self.auth = hs.get_v1auth() - self.txns = HttpTransactionCache() + self.txns = HttpTransactionCache(hs.get_clock()) diff --git a/synapse/rest/client/v2_alpha/sendtodevice.py b/synapse/rest/client/v2_alpha/sendtodevice.py index 2187350d42..ac660669f3 100644 --- a/synapse/rest/client/v2_alpha/sendtodevice.py +++ b/synapse/rest/client/v2_alpha/sendtodevice.py @@ -40,7 +40,7 @@ class SendToDeviceRestServlet(servlet.RestServlet): super(SendToDeviceRestServlet, self).__init__() self.hs = hs self.auth = hs.get_auth() - self.txns = HttpTransactionCache() + self.txns = HttpTransactionCache(hs.get_clock()) self.device_message_handler = hs.get_device_message_handler() def on_PUT(self, request, message_type, txn_id): diff --git a/synapse/util/__init__.py b/synapse/util/__init__.py index 2b3f0bef3c..c05b9450be 100644 --- a/synapse/util/__init__.py +++ b/synapse/util/__init__.py @@ -34,7 +34,7 @@ class Clock(object): """A small utility that obtains current time-of-day so that time may be mocked during unit-tests. - TODO(paul): Also move the sleep() functionallity into it + TODO(paul): Also move the sleep() functionality into it """ def time(self): @@ -46,6 +46,14 @@ class Clock(object): return int(self.time() * 1000) def looping_call(self, f, msec): + """Call a function repeatedly. + + Waits `msec` initially before calling `f` for the first time. + + Args: + f(function): The function to call repeatedly. + msec(float): How long to wait between calls in milliseconds. + """ l = task.LoopingCall(f) l.start(msec / 1000.0, now=False) return l diff --git a/tests/rest/client/test_transactions.py b/tests/rest/client/test_transactions.py new file mode 100644 index 0000000000..d7cea30260 --- /dev/null +++ b/tests/rest/client/test_transactions.py @@ -0,0 +1,69 @@ +from synapse.rest.client.transactions import HttpTransactionCache +from synapse.rest.client.transactions import CLEANUP_PERIOD_MS +from twisted.internet import defer +from mock import Mock, call +from tests import unittest +from tests.utils import MockClock + + +class HttpTransactionCacheTestCase(unittest.TestCase): + + def setUp(self): + self.clock = MockClock() + self.cache = HttpTransactionCache(self.clock) + + self.mock_http_response = (200, "GOOD JOB!") + self.mock_key = "foo" + + @defer.inlineCallbacks + def test_executes_given_function(self): + cb = Mock( + return_value=defer.succeed(self.mock_http_response) + ) + res = yield self.cache.fetch_or_execute( + self.mock_key, cb, "some_arg", keyword="arg" + ) + cb.assert_called_once_with("some_arg", keyword="arg") + self.assertEqual(res, self.mock_http_response) + + @defer.inlineCallbacks + def test_deduplicates_based_on_key(self): + cb = Mock( + return_value=defer.succeed(self.mock_http_response) + ) + for i in range(3): # invoke multiple times + res = yield self.cache.fetch_or_execute( + self.mock_key, cb, "some_arg", keyword="arg", changing_args=i + ) + self.assertEqual(res, self.mock_http_response) + # expect only a single call to do the work + cb.assert_called_once_with("some_arg", keyword="arg", changing_args=0) + + @defer.inlineCallbacks + def test_cleans_up(self): + cb = Mock( + return_value=defer.succeed(self.mock_http_response) + ) + yield self.cache.fetch_or_execute( + self.mock_key, cb, "an arg" + ) + # should NOT have cleaned up yet + self.clock.advance_time_msec(CLEANUP_PERIOD_MS / 2) + + yield self.cache.fetch_or_execute( + self.mock_key, cb, "an arg" + ) + # still using cache + cb.assert_called_once_with("an arg") + + self.clock.advance_time_msec(CLEANUP_PERIOD_MS) + + yield self.cache.fetch_or_execute( + self.mock_key, cb, "an arg" + ) + # no longer using cache + self.assertEqual(cb.call_count, 2) + self.assertEqual( + cb.call_args_list, + [call("an arg",), call("an arg",)] + )