Clean transactions based on time. Add HttpTransactionCache tests.

4 files changed, 30 insertions, 8 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]
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