We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Split fed worker & sender config means destination retry sched doesn't invalidate when you receive traffic


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `synapse/storage/transactions.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 # Copyright 2014-2016 OpenMarket Ltd
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 #     http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 
16 import logging
17 from collections import namedtuple
18 
19 import six
20 
21 from canonicaljson import encode_canonical_json
22 
23 from twisted.internet import defer
24 
25 from synapse.metrics.background_process_metrics import run_as_background_process
26 from synapse.util.caches.descriptors import cached
27 
28 from ._base import SQLBaseStore, db_to_json
29 
30 # py2 sqlite has buffer hardcoded as only binary type, so we must use it,
31 # despite being deprecated and removed in favor of memoryview
32 if six.PY2:
33     db_binary_type = buffer
34 else:
35     db_binary_type = memoryview
36 
37 logger = logging.getLogger(__name__)
38 
39 
40 _TransactionRow = namedtuple(
41     "_TransactionRow", (
42         "id", "transaction_id", "destination", "ts", "response_code",
43         "response_json",
44     )
45 )
46 
47 _UpdateTransactionRow = namedtuple(
48     "_TransactionRow", (
49         "response_code", "response_json",
50     )
51 )
52 
53 
54 class TransactionStore(SQLBaseStore):
55     """A collection of queries for handling PDUs.
56     """
57 
58     def __init__(self, db_conn, hs):
59         super(TransactionStore, self).__init__(db_conn, hs)
60 
61         self._clock.looping_call(self._start_cleanup_transactions, 30 * 60 * 1000)
62 
63     def get_received_txn_response(self, transaction_id, origin):
64         """For an incoming transaction from a given origin, check if we have
65         already responded to it. If so, return the response code and response
66         body (as a dict).
67 
68         Args:
69             transaction_id (str)
70             origin(str)
71 
72         Returns:
73             tuple: None if we have not previously responded to
74             this transaction or a 2-tuple of (int, dict)
75         """
76 
77         return self.runInteraction(
78             "get_received_txn_response",
79             self._get_received_txn_response, transaction_id, origin
80         )
81 
82     def _get_received_txn_response(self, txn, transaction_id, origin):
83         result = self._simple_select_one_txn(
84             txn,
85             table="received_transactions",
86             keyvalues={
87                 "transaction_id": transaction_id,
88                 "origin": origin,
89             },
90             retcols=(
91                 "transaction_id", "origin", "ts", "response_code", "response_json",
92                 "has_been_referenced",
93             ),
94             allow_none=True,
95         )
96 
97         if result and result["response_code"]:
98             return result["response_code"], db_to_json(result["response_json"])
99 
100         else:
101             return None
102 
103     def set_received_txn_response(self, transaction_id, origin, code,
104                                   response_dict):
105         """Persist the response we returened for an incoming transaction, and
106         should return for subsequent transactions with the same transaction_id
107         and origin.
108 
109         Args:
110             txn
111             transaction_id (str)
112             origin (str)
113             code (int)
114             response_json (str)
115         """
116 
117         return self._simple_insert(
118             table="received_transactions",
119             values={
120                 "transaction_id": transaction_id,
121                 "origin": origin,
122                 "response_code": code,
123                 "response_json": db_binary_type(encode_canonical_json(response_dict)),
124                 "ts": self._clock.time_msec(),
125             },
126             or_ignore=True,
127             desc="set_received_txn_response",
128         )
129 
130     def prep_send_transaction(self, transaction_id, destination,
131                               origin_server_ts):
132         """Persists an outgoing transaction and calculates the values for the
133         previous transaction id list.
134 
135         This should be called before sending the transaction so that it has the
136         correct value for the `prev_ids` key.
137 
138         Args:
139             transaction_id (str)
140             destination (str)
141             origin_server_ts (int)
142 
143         Returns:
144             list: A list of previous transaction ids.
145         """
146         return defer.succeed([])
147 
148     def delivered_txn(self, transaction_id, destination, code, response_dict):
149         """Persists the response for an outgoing transaction.
150 
151         Args:
152             transaction_id (str)
153             destination (str)
154             code (int)
155             response_json (str)
156         """
157         pass
158 
159     @cached(max_entries=10000)
160     def get_destination_retry_timings(self, destination):
161         """Gets the current retry timings (if any) for a given destination.
162 
163         Args:
164             destination (str)
165 
166         Returns:
167             None if not retrying
168             Otherwise a dict for the retry scheme
169         """
170         return self.runInteraction(
171             "get_destination_retry_timings",
172             self._get_destination_retry_timings, destination)
173 
174     def _get_destination_retry_timings(self, txn, destination):
175         result = self._simple_select_one_txn(
176             txn,
177             table="destinations",
178             keyvalues={
179                 "destination": destination,
180             },
181             retcols=("destination", "retry_last_ts", "retry_interval"),
182             allow_none=True,
183         )
184 
185         if result and result["retry_last_ts"] > 0:
186             return result
187         else:
188             return None
189 
190     def set_destination_retry_timings(self, destination,
191                                       retry_last_ts, retry_interval):
192         """Sets the current retry timings for a given destination.
193         Both timings should be zero if retrying is no longer occuring.
194 
195         Args:
196             destination (str)
197             retry_last_ts (int) - time of last retry attempt in unix epoch ms
198             retry_interval (int) - how long until next retry in ms
199         """
200 
201         # XXX: we could chose to not bother persisting this if our cache thinks
202         # this is a NOOP
203         return self.runInteraction(
204             "set_destination_retry_timings",
205             self._set_destination_retry_timings,
206             destination,
207             retry_last_ts,
208             retry_interval,
209         )
210 
211     def _set_destination_retry_timings(self, txn, destination,
212                                        retry_last_ts, retry_interval):
213         self.database_engine.lock_table(txn, "destinations")
214 
215         self._invalidate_cache_and_stream(
216             txn, self.get_destination_retry_timings, (destination,)
217         )
218 
219         # We need to be careful here as the data may have changed from under us
220         # due to a worker setting the timings.
221 
222         prev_row = self._simple_select_one_txn(
223             txn,
224             table="destinations",
225             keyvalues={
226                 "destination": destination,
227             },
228             retcols=("retry_last_ts", "retry_interval"),
229             allow_none=True,
230         )
231 
232         if not prev_row:
233             self._simple_insert_txn(
234                 txn,
235                 table="destinations",
236                 values={
237                     "destination": destination,
238                     "retry_last_ts": retry_last_ts,
239                     "retry_interval": retry_interval,
240                 }
241             )
242         elif retry_interval == 0 or prev_row["retry_interval"] < retry_interval:
243             self._simple_update_one_txn(
244                 txn,
245                 "destinations",
246                 keyvalues={
247                     "destination": destination,
248                 },
249                 updatevalues={
250                     "retry_last_ts": retry_last_ts,
251                     "retry_interval": retry_interval,
252                 },
253             )
254 
255     def get_destinations_needing_retry(self):
256         """Get all destinations which are due a retry for sending a transaction.
257 
258         Returns:
259             list: A list of dicts
260         """
261 
262         return self.runInteraction(
263             "get_destinations_needing_retry",
264             self._get_destinations_needing_retry
265         )
266 
267     def _get_destinations_needing_retry(self, txn):
268         query = (
269             "SELECT * FROM destinations"
270             " WHERE retry_last_ts > 0 and retry_next_ts < ?"
271         )
272 
273         txn.execute(query, (self._clock.time_msec(),))
274         return self.cursor_to_dict(txn)
275 
276     def _start_cleanup_transactions(self):
277         return run_as_background_process(
278             "cleanup_transactions", self._cleanup_transactions,
279         )
280 
281     def _cleanup_transactions(self):
282         now = self._clock.time_msec()
283         month_ago = now - 30 * 24 * 60 * 60 * 1000
284 
285         def _cleanup_transactions_txn(txn):
286             txn.execute("DELETE FROM received_transactions WHERE ts < ?", (month_ago,))
287 
288         return self.runInteraction("_cleanup_transactions", _cleanup_transactions_txn)
289
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/synapse/storage/transactions.py b/synapse/storage/transactions.py
--- a/synapse/storage/transactions.py
+++ b/synapse/storage/transactions.py
@@ -23,7 +23,6 @@
 from twisted.internet import defer
 
 from synapse.metrics.background_process_metrics import run_as_background_process
-from synapse.util.caches.descriptors import cached
 
 from ._base import SQLBaseStore, db_to_json
 
@@ -156,7 +155,6 @@
         """
         pass
 
-    @cached(max_entries=10000)
     def get_destination_retry_timings(self, destination):
         """Gets the current retry timings (if any) for a given destination.
 
@@ -198,8 +196,6 @@
             retry_interval (int) - how long until next retry in ms
         """
 
-        # XXX: we could chose to not bother persisting this if our cache thinks
-        # this is a NOOP
         return self.runInteraction(
             "set_destination_retry_timings",
             self._set_destination_retry_timings,
@@ -212,10 +208,6 @@
                                        retry_last_ts, retry_interval):
         self.database_engine.lock_table(txn, "destinations")
 
-        self._invalidate_cache_and_stream(
-            txn, self.get_destination_retry_timings, (destination,)
-        )
-
         # We need to be careful here as the data may have changed from under us
         # due to a worker setting the timings.

{"golden_diff": "diff --git a/synapse/storage/transactions.py b/synapse/storage/transactions.py\n--- a/synapse/storage/transactions.py\n+++ b/synapse/storage/transactions.py\n@@ -23,7 +23,6 @@\n from twisted.internet import defer\n \n from synapse.metrics.background_process_metrics import run_as_background_process\n-from synapse.util.caches.descriptors import cached\n \n from ._base import SQLBaseStore, db_to_json\n \n@@ -156,7 +155,6 @@\n         \"\"\"\n         pass\n \n-    @cached(max_entries=10000)\n     def get_destination_retry_timings(self, destination):\n         \"\"\"Gets the current retry timings (if any) for a given destination.\n \n@@ -198,8 +196,6 @@\n             retry_interval (int) - how long until next retry in ms\n         \"\"\"\n \n-        # XXX: we could chose to not bother persisting this if our cache thinks\n-        # this is a NOOP\n         return self.runInteraction(\n             \"set_destination_retry_timings\",\n             self._set_destination_retry_timings,\n@@ -212,10 +208,6 @@\n                                        retry_last_ts, retry_interval):\n         self.database_engine.lock_table(txn, \"destinations\")\n \n-        self._invalidate_cache_and_stream(\n-            txn, self.get_destination_retry_timings, (destination,)\n-        )\n-\n         # We need to be careful here as the data may have changed from under us\n         # due to a worker setting the timings.\n", "issue": "Split fed worker & sender config means destination retry sched doesn't invalidate when you receive traffic\n\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n# Copyright 2014-2016 OpenMarket Ltd\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport logging\nfrom collections import namedtuple\n\nimport six\n\nfrom canonicaljson import encode_canonical_json\n\nfrom twisted.internet import defer\n\nfrom synapse.metrics.background_process_metrics import run_as_background_process\nfrom synapse.util.caches.descriptors import cached\n\nfrom ._base import SQLBaseStore, db_to_json\n\n# py2 sqlite has buffer hardcoded as only binary type, so we must use it,\n# despite being deprecated and removed in favor of memoryview\nif six.PY2:\n    db_binary_type = buffer\nelse:\n    db_binary_type = memoryview\n\nlogger = logging.getLogger(__name__)\n\n\n_TransactionRow = namedtuple(\n    \"_TransactionRow\", (\n        \"id\", \"transaction_id\", \"destination\", \"ts\", \"response_code\",\n        \"response_json\",\n    )\n)\n\n_UpdateTransactionRow = namedtuple(\n    \"_TransactionRow\", (\n        \"response_code\", \"response_json\",\n    )\n)\n\n\nclass TransactionStore(SQLBaseStore):\n    \"\"\"A collection of queries for handling PDUs.\n    \"\"\"\n\n    def __init__(self, db_conn, hs):\n        super(TransactionStore, self).__init__(db_conn, hs)\n\n        self._clock.looping_call(self._start_cleanup_transactions, 30 * 60 * 1000)\n\n    def get_received_txn_response(self, transaction_id, origin):\n        \"\"\"For an incoming transaction from a given origin, check if we have\n        already responded to it. If so, return the response code and response\n        body (as a dict).\n\n        Args:\n            transaction_id (str)\n            origin(str)\n\n        Returns:\n            tuple: None if we have not previously responded to\n            this transaction or a 2-tuple of (int, dict)\n        \"\"\"\n\n        return self.runInteraction(\n            \"get_received_txn_response\",\n            self._get_received_txn_response, transaction_id, origin\n        )\n\n    def _get_received_txn_response(self, txn, transaction_id, origin):\n        result = self._simple_select_one_txn(\n            txn,\n            table=\"received_transactions\",\n            keyvalues={\n                \"transaction_id\": transaction_id,\n                \"origin\": origin,\n            },\n            retcols=(\n                \"transaction_id\", \"origin\", \"ts\", \"response_code\", \"response_json\",\n                \"has_been_referenced\",\n            ),\n            allow_none=True,\n        )\n\n        if result and result[\"response_code\"]:\n            return result[\"response_code\"], db_to_json(result[\"response_json\"])\n\n        else:\n            return None\n\n    def set_received_txn_response(self, transaction_id, origin, code,\n                                  response_dict):\n        \"\"\"Persist the response we returened for an incoming transaction, and\n        should return for subsequent transactions with the same transaction_id\n        and origin.\n\n        Args:\n            txn\n            transaction_id (str)\n            origin (str)\n            code (int)\n            response_json (str)\n        \"\"\"\n\n        return self._simple_insert(\n            table=\"received_transactions\",\n            values={\n                \"transaction_id\": transaction_id,\n                \"origin\": origin,\n                \"response_code\": code,\n                \"response_json\": db_binary_type(encode_canonical_json(response_dict)),\n                \"ts\": self._clock.time_msec(),\n            },\n            or_ignore=True,\n            desc=\"set_received_txn_response\",\n        )\n\n    def prep_send_transaction(self, transaction_id, destination,\n                              origin_server_ts):\n        \"\"\"Persists an outgoing transaction and calculates the values for the\n        previous transaction id list.\n\n        This should be called before sending the transaction so that it has the\n        correct value for the `prev_ids` key.\n\n        Args:\n            transaction_id (str)\n            destination (str)\n            origin_server_ts (int)\n\n        Returns:\n            list: A list of previous transaction ids.\n        \"\"\"\n        return defer.succeed([])\n\n    def delivered_txn(self, transaction_id, destination, code, response_dict):\n        \"\"\"Persists the response for an outgoing transaction.\n\n        Args:\n            transaction_id (str)\n            destination (str)\n            code (int)\n            response_json (str)\n        \"\"\"\n        pass\n\n    @cached(max_entries=10000)\n    def get_destination_retry_timings(self, destination):\n        \"\"\"Gets the current retry timings (if any) for a given destination.\n\n        Args:\n            destination (str)\n\n        Returns:\n            None if not retrying\n            Otherwise a dict for the retry scheme\n        \"\"\"\n        return self.runInteraction(\n            \"get_destination_retry_timings\",\n            self._get_destination_retry_timings, destination)\n\n    def _get_destination_retry_timings(self, txn, destination):\n        result = self._simple_select_one_txn(\n            txn,\n            table=\"destinations\",\n            keyvalues={\n                \"destination\": destination,\n            },\n            retcols=(\"destination\", \"retry_last_ts\", \"retry_interval\"),\n            allow_none=True,\n        )\n\n        if result and result[\"retry_last_ts\"] > 0:\n            return result\n        else:\n            return None\n\n    def set_destination_retry_timings(self, destination,\n                                      retry_last_ts, retry_interval):\n        \"\"\"Sets the current retry timings for a given destination.\n        Both timings should be zero if retrying is no longer occuring.\n\n        Args:\n            destination (str)\n            retry_last_ts (int) - time of last retry attempt in unix epoch ms\n            retry_interval (int) - how long until next retry in ms\n        \"\"\"\n\n        # XXX: we could chose to not bother persisting this if our cache thinks\n        # this is a NOOP\n        return self.runInteraction(\n            \"set_destination_retry_timings\",\n            self._set_destination_retry_timings,\n            destination,\n            retry_last_ts,\n            retry_interval,\n        )\n\n    def _set_destination_retry_timings(self, txn, destination,\n                                       retry_last_ts, retry_interval):\n        self.database_engine.lock_table(txn, \"destinations\")\n\n        self._invalidate_cache_and_stream(\n            txn, self.get_destination_retry_timings, (destination,)\n        )\n\n        # We need to be careful here as the data may have changed from under us\n        # due to a worker setting the timings.\n\n        prev_row = self._simple_select_one_txn(\n            txn,\n            table=\"destinations\",\n            keyvalues={\n                \"destination\": destination,\n            },\n            retcols=(\"retry_last_ts\", \"retry_interval\"),\n            allow_none=True,\n        )\n\n        if not prev_row:\n            self._simple_insert_txn(\n                txn,\n                table=\"destinations\",\n                values={\n                    \"destination\": destination,\n                    \"retry_last_ts\": retry_last_ts,\n                    \"retry_interval\": retry_interval,\n                }\n            )\n        elif retry_interval == 0 or prev_row[\"retry_interval\"] < retry_interval:\n            self._simple_update_one_txn(\n                txn,\n                \"destinations\",\n                keyvalues={\n                    \"destination\": destination,\n                },\n                updatevalues={\n                    \"retry_last_ts\": retry_last_ts,\n                    \"retry_interval\": retry_interval,\n                },\n            )\n\n    def get_destinations_needing_retry(self):\n        \"\"\"Get all destinations which are due a retry for sending a transaction.\n\n        Returns:\n            list: A list of dicts\n        \"\"\"\n\n        return self.runInteraction(\n            \"get_destinations_needing_retry\",\n            self._get_destinations_needing_retry\n        )\n\n    def _get_destinations_needing_retry(self, txn):\n        query = (\n            \"SELECT * FROM destinations\"\n            \" WHERE retry_last_ts > 0 and retry_next_ts < ?\"\n        )\n\n        txn.execute(query, (self._clock.time_msec(),))\n        return self.cursor_to_dict(txn)\n\n    def _start_cleanup_transactions(self):\n        return run_as_background_process(\n            \"cleanup_transactions\", self._cleanup_transactions,\n        )\n\n    def _cleanup_transactions(self):\n        now = self._clock.time_msec()\n        month_ago = now - 30 * 24 * 60 * 60 * 1000\n\n        def _cleanup_transactions_txn(txn):\n            txn.execute(\"DELETE FROM received_transactions WHERE ts < ?\", (month_ago,))\n\n        return self.runInteraction(\"_cleanup_transactions\", _cleanup_transactions_txn)\n", "path": "synapse/storage/transactions.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n# Copyright 2014-2016 OpenMarket Ltd\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport logging\nfrom collections import namedtuple\n\nimport six\n\nfrom canonicaljson import encode_canonical_json\n\nfrom twisted.internet import defer\n\nfrom synapse.metrics.background_process_metrics import run_as_background_process\n\nfrom ._base import SQLBaseStore, db_to_json\n\n# py2 sqlite has buffer hardcoded as only binary type, so we must use it,\n# despite being deprecated and removed in favor of memoryview\nif six.PY2:\n    db_binary_type = buffer\nelse:\n    db_binary_type = memoryview\n\nlogger = logging.getLogger(__name__)\n\n\n_TransactionRow = namedtuple(\n    \"_TransactionRow\", (\n        \"id\", \"transaction_id\", \"destination\", \"ts\", \"response_code\",\n        \"response_json\",\n    )\n)\n\n_UpdateTransactionRow = namedtuple(\n    \"_TransactionRow\", (\n        \"response_code\", \"response_json\",\n    )\n)\n\n\nclass TransactionStore(SQLBaseStore):\n    \"\"\"A collection of queries for handling PDUs.\n    \"\"\"\n\n    def __init__(self, db_conn, hs):\n        super(TransactionStore, self).__init__(db_conn, hs)\n\n        self._clock.looping_call(self._start_cleanup_transactions, 30 * 60 * 1000)\n\n    def get_received_txn_response(self, transaction_id, origin):\n        \"\"\"For an incoming transaction from a given origin, check if we have\n        already responded to it. If so, return the response code and response\n        body (as a dict).\n\n        Args:\n            transaction_id (str)\n            origin(str)\n\n        Returns:\n            tuple: None if we have not previously responded to\n            this transaction or a 2-tuple of (int, dict)\n        \"\"\"\n\n        return self.runInteraction(\n            \"get_received_txn_response\",\n            self._get_received_txn_response, transaction_id, origin\n        )\n\n    def _get_received_txn_response(self, txn, transaction_id, origin):\n        result = self._simple_select_one_txn(\n            txn,\n            table=\"received_transactions\",\n            keyvalues={\n                \"transaction_id\": transaction_id,\n                \"origin\": origin,\n            },\n            retcols=(\n                \"transaction_id\", \"origin\", \"ts\", \"response_code\", \"response_json\",\n                \"has_been_referenced\",\n            ),\n            allow_none=True,\n        )\n\n        if result and result[\"response_code\"]:\n            return result[\"response_code\"], db_to_json(result[\"response_json\"])\n\n        else:\n            return None\n\n    def set_received_txn_response(self, transaction_id, origin, code,\n                                  response_dict):\n        \"\"\"Persist the response we returened for an incoming transaction, and\n        should return for subsequent transactions with the same transaction_id\n        and origin.\n\n        Args:\n            txn\n            transaction_id (str)\n            origin (str)\n            code (int)\n            response_json (str)\n        \"\"\"\n\n        return self._simple_insert(\n            table=\"received_transactions\",\n            values={\n                \"transaction_id\": transaction_id,\n                \"origin\": origin,\n                \"response_code\": code,\n                \"response_json\": db_binary_type(encode_canonical_json(response_dict)),\n                \"ts\": self._clock.time_msec(),\n            },\n            or_ignore=True,\n            desc=\"set_received_txn_response\",\n        )\n\n    def prep_send_transaction(self, transaction_id, destination,\n                              origin_server_ts):\n        \"\"\"Persists an outgoing transaction and calculates the values for the\n        previous transaction id list.\n\n        This should be called before sending the transaction so that it has the\n        correct value for the `prev_ids` key.\n\n        Args:\n            transaction_id (str)\n            destination (str)\n            origin_server_ts (int)\n\n        Returns:\n            list: A list of previous transaction ids.\n        \"\"\"\n        return defer.succeed([])\n\n    def delivered_txn(self, transaction_id, destination, code, response_dict):\n        \"\"\"Persists the response for an outgoing transaction.\n\n        Args:\n            transaction_id (str)\n            destination (str)\n            code (int)\n            response_json (str)\n        \"\"\"\n        pass\n\n    def get_destination_retry_timings(self, destination):\n        \"\"\"Gets the current retry timings (if any) for a given destination.\n\n        Args:\n            destination (str)\n\n        Returns:\n            None if not retrying\n            Otherwise a dict for the retry scheme\n        \"\"\"\n        return self.runInteraction(\n            \"get_destination_retry_timings\",\n            self._get_destination_retry_timings, destination)\n\n    def _get_destination_retry_timings(self, txn, destination):\n        result = self._simple_select_one_txn(\n            txn,\n            table=\"destinations\",\n            keyvalues={\n                \"destination\": destination,\n            },\n            retcols=(\"destination\", \"retry_last_ts\", \"retry_interval\"),\n            allow_none=True,\n        )\n\n        if result and result[\"retry_last_ts\"] > 0:\n            return result\n        else:\n            return None\n\n    def set_destination_retry_timings(self, destination,\n                                      retry_last_ts, retry_interval):\n        \"\"\"Sets the current retry timings for a given destination.\n        Both timings should be zero if retrying is no longer occuring.\n\n        Args:\n            destination (str)\n            retry_last_ts (int) - time of last retry attempt in unix epoch ms\n            retry_interval (int) - how long until next retry in ms\n        \"\"\"\n\n        return self.runInteraction(\n            \"set_destination_retry_timings\",\n            self._set_destination_retry_timings,\n            destination,\n            retry_last_ts,\n            retry_interval,\n        )\n\n    def _set_destination_retry_timings(self, txn, destination,\n                                       retry_last_ts, retry_interval):\n        self.database_engine.lock_table(txn, \"destinations\")\n\n        # We need to be careful here as the data may have changed from under us\n        # due to a worker setting the timings.\n\n        prev_row = self._simple_select_one_txn(\n            txn,\n            table=\"destinations\",\n            keyvalues={\n                \"destination\": destination,\n            },\n            retcols=(\"retry_last_ts\", \"retry_interval\"),\n            allow_none=True,\n        )\n\n        if not prev_row:\n            self._simple_insert_txn(\n                txn,\n                table=\"destinations\",\n                values={\n                    \"destination\": destination,\n                    \"retry_last_ts\": retry_last_ts,\n                    \"retry_interval\": retry_interval,\n                }\n            )\n        elif retry_interval == 0 or prev_row[\"retry_interval\"] < retry_interval:\n            self._simple_update_one_txn(\n                txn,\n                \"destinations\",\n                keyvalues={\n                    \"destination\": destination,\n                },\n                updatevalues={\n                    \"retry_last_ts\": retry_last_ts,\n                    \"retry_interval\": retry_interval,\n                },\n            )\n\n    def get_destinations_needing_retry(self):\n        \"\"\"Get all destinations which are due a retry for sending a transaction.\n\n        Returns:\n            list: A list of dicts\n        \"\"\"\n\n        return self.runInteraction(\n            \"get_destinations_needing_retry\",\n            self._get_destinations_needing_retry\n        )\n\n    def _get_destinations_needing_retry(self, txn):\n        query = (\n            \"SELECT * FROM destinations\"\n            \" WHERE retry_last_ts > 0 and retry_next_ts < ?\"\n        )\n\n        txn.execute(query, (self._clock.time_msec(),))\n        return self.cursor_to_dict(txn)\n\n    def _start_cleanup_transactions(self):\n        return run_as_background_process(\n            \"cleanup_transactions\", self._cleanup_transactions,\n        )\n\n    def _cleanup_transactions(self):\n        now = self._clock.time_msec()\n        month_ago = now - 30 * 24 * 60 * 60 * 1000\n\n        def _cleanup_transactions_txn(txn):\n            txn.execute(\"DELETE FROM received_transactions WHERE ts < ?\", (month_ago,))\n\n        return self.runInteraction(\"_cleanup_transactions\", _cleanup_transactions_txn)\n", "path": "synapse/storage/transactions.py"}]}