diff options
Diffstat (limited to 'synapse/rest/client/transactions.py')
| -rw-r--r-- | synapse/rest/client/transactions.py | 24 |
1 files changed, 19 insertions, 5 deletions
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] |